使用MapReduce运行WordCount案例

public class WCMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
	
	private Text out_key=new Text();
	private IntWritable out_value=new IntWritable(1);//每个单词出现一次记为1
	
	// 针对输入的每个 keyin-valuein调用一次   (0，hello	hi	hello	hi)
	@Override
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws Exception {
	
		System.out.println("keyin:"+key+"----keyout:"+value);
		
		String[] words = value.toString().split("\t");
		
		for (String word : words) {
			out_key.set(word);
			//写出数据(单词,1)
			context.write(out_key, out_value);
		}
			
	}
}

/* 
 *KEYIN, VALUEIN: Mapper输出的keyout-valueout
 *KEYOUT, VALUEOUT: 自定义		
 */		
public class WCReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
	
	private IntWritable out_value=new IntWritable();
	
	// reduce一次处理一组数据，key相同的视为一组
	@Override
	protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws Exception {
		
		int sum=0;
		
		for (IntWritable intWritable : values) {
			sum += intWritable.get();
		}
		
		out_value.set(sum);
		
		//将累加的值写出
		context.write(key, out_value);
		
	}
}

/*
 * 1.启动这个线程，运行Job
 * 
 * 2.本地模式主要用于测试程序是否正确！	
 */
public class WCDriver {
	
	public static void main(String[] args) throws Exception {
		
		Path inputPath=new Path("e:/input/wordcount");
		Path outputPath=new Path("e:/output/wordcount");//保证输出目录不存在
		
		//作为整个Job的配置
		Configuration conf = new Configuration();//空参表示默认使用本地的文件系统
		
		//使用HDFS，分布式文件系统
		/*
		Path inputPath=new Path("/wordcount");
		Path outputPath=new Path("/mroutput/wordcount");
		
		conf.set("fs.defaultFS", "hdfs://hadoop101:9000");
		
		conf.set("mapreduce.framework.name", "yarn");// 在YARN上运行
		
		conf.set("yarn.resourcemanager.hostname", "hadoop102");// RM所在的机器
		*/
		
		//一定要保证输出目录不存在
		FileSystem fs=FileSystem.get(conf);
		
		if (fs.exists(outputPath)) {
			fs.delete(outputPath, true);
		}
		
		// ①创建Job
		Job job = Job.getInstance(conf);

		//job.setJar("MapReduce-0.0.1-SNAPSHOT.jar");// 告诉NM运行时，MR中Job所在的Jar包在哪里
		
		// 将某个类所在地jar包作为job的jar包
		job.setJarByClass(WCDriver.class);
			
		// 为Job创建一个名字
		job.setJobName("wordcount");
		
		// ②设置Job
		// 设置Job运行的Mapper，Reducer类型，Mapper,Reducer输出的key-value类型
		job.setMapperClass(WCMapper.class);
		job.setReducerClass(WCReducer.class);
		
		// Job需要根据Mapper和Reducer输出的Key-value类型准备序列化器，通过序列化器对输出的key-value进行序列化和反序列化
		// 如果Mapper和Reducer输出的Key-value类型一致，直接设置Job最终的输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		// 设置输入目录和输出目录
		FileInputFormat.setInputPaths(job, inputPath);
		FileOutputFormat.setOutputPath(job, outputPath);
		
		// ③运行Job
		job.waitForCompletion(true);
			
	}
}

hadoop jar jar包名 主类名(WCDriver)