大数据 （010）Hadoop-第一次使用eclipse&java&hadoop分析简单数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
package com.zjt.mapreducer.data;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.StringUtils;

/**
 * 执行统计单词数量的map程序
 * @author ZhangJintao
 * 		Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>
 * 			KEYIN ----   输入数据的键
 * 			VALUEIN ----  输入数据的值
 * 			KEYOUT ---- 输出数据的键
 * 			VALUEOUT ----  输出数据的值
 */
public class WorldCounteMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
	@Override
	/**
	 * 父类的map方法，循环调用
	 * 从split碎片段中每行调用一次
	 * 把该行所在下标为key，该行的值为value
	 * 【功能：将单词以map输出】
	 */
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
			throws IOException, InterruptedException {
		String[] words = StringUtils.split(value.toString(), ' ');
		for (String w : words) {
			context.write(new Text(w), new IntWritable(1));
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
package com.zjt.mapreducer.data;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class WorldCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
	@Override
	/**
	 * 循环调用
	 * 洗牌完毕分好组后，每组调用一次
	 * 【功能：计算单词出现次数】
	 */
	protected void reduce(Text arg0, Iterable<IntWritable> arg1,
			Reducer<Text, IntWritable, Text, IntWritable>.Context arg2) throws IOException, InterruptedException {
		int sum = 0 ;
		for(IntWritable i : arg1){
			sum += i.get();
		}
		arg2.write(arg0, new IntWritable(sum));
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
package com.zjt.mapreducer.data;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 执行方法
 * @author ZhangJintao
 */
public class RunJob {
	public static void main(String[] args) {
		Configuration config = new Configuration();
		try {
			FileSystem fs = FileSystem.get(config);
			
			Job job = Job.getInstance();
			//执行mapreducer任务
			job.setJarByClass(RunJob.class);
			job.setJobName("data");
			
			job.setMapperClass(WorldCounteMapper.class);
			job.setReducerClass(WorldCountReducer.class);
			
			job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(IntWritable.class);
			
			FileInputFormat.addInputPath(job, new Path("/usr/input/"));
			Path outpath  = new Path("/usr/input/data");
			if (fs.exists(outpath)) {
				fs.delete(outpath, true);
			}
			FileOutputFormat.setOutputPath(job, outpath);
			
			boolean f = job.waitForCompletion(true);
			
			if (f) {
				System.out.println("JOB 执行成功");
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}

1
2
3
hadoop hello world
hello hadoop
hbase zookeeper

1
    hadoop jar wc.jar com.zjt.mapreducer.data.RunJob