MapReduce 的 詞語統計 WordCount 統計

創建一個包：cn.itcast.hadoop.mr.wordcount。

三個類：WCmapper、WCReducer、WCRunner。

 

WCmapper

package cn.itcast.hadoop.mr.wordcount;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper; //包沒有導進來，硬着頭皮看

//4個泛型中，前兩個是指定 mapper 輸入數據的類型，keyin 是輸入的 key 類型，valuein 是輸入的 value 類型
//map 和 reduce 的數據輸入輸出都是以 key-value 形式封裝的
//默認狀況下，框架傳遞給咱們的 mapper 的輸入數據中，key 是要處理的文本一行的起始偏移量，這一行的內容做爲 value
public class WCmapper extends Mapper<Long, String, String, Long> {

	// mapreduce框架每讀一行數據就調用一次該方法
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		// 具體業務邏輯就寫在這個方法體中，並且咱們業務處理的數據已經被框架傳遞進來，在方法的參數中key-value
		// key 是這一行數據的其實偏移量，value 是這一行的文本內容

		// 將這一行的內容轉化爲 String類型
		String line = value.toString();

		// 對這一行的文本按照特定分隔符切分
		String[] words = StringUtils.split(line, " ");

		// 遍歷這個單詞數組輸出爲 kv 形式，k：單詞；v：1
		for (String word : words) {

			context.write(new Text(word), new LongWritable(1));

		}

	}

}

WCReducer

package cn.itcast.hadoop.mr.wordcount;

import java.io.IOException;

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

public class WCReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

	// 框架在 map 處理完成以後，將對全部kv緩存起來，進行分組，而後傳遞一個組<key,value{}>,調用一次reduce方法
	// <hello,{1,1,1,1,1,1.....}>
	@Override
	protected void reduce(Text key, Iterable<LongWritable> values, Context context)
			throws IOException, InterruptedException {

		long count = 0;
		// 遍歷value的list，進行累加求和
		for (LongWritable value : values) {

			count += value.get();
		}

		// 輸出一個單詞的統計結果

		context.write(key, new LongWritable(count));

	}

}

WCRunner

package cn.itcast.hadoop.mr.wordcount;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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;

/**
 * 用來描述一個特定的類 好比，該做業使用哪一個類做爲邏輯處理中的map，哪一個做爲MapReduce 開能夠指定該做爲要處理的數據所在的路徑
 * 還能夠指定該做業輸出的結果放到哪一個路徑 ....
 * 
 * @author duanhaitao@itcast.cn
 *
 */
public class WCRunner {

	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();

		Job wcjob = Job.getInstance(conf);

		// 設置整個job所用類在哪一個jar包
		wcjob.setJarByClass(WCRunner.class);

		// 本 job 使用的 mapper 和reducer的類
		wcjob.setMapperClass(WCMapper.class);
		wcjob.setReducerClass(WCReducer.class);

		// ָ指定 reduce 的輸出數據kv類型
		wcjob.setOutputKeyClass(Text.class);
		wcjob.setOutputValueClass(LongWritable.class);

		// 指定 mapper 的輸出數據kv類型
		wcjob.setMapOutputKeyClass(Text.class);
		wcjob.setMapOutputValueClass(LongWritable.class);

		// 原始數據存放路徑
		FileInputFormat.setInputPaths(wcjob, new Path("hdfs://weekend101:9000/wc/srcdata/"));

		// ָ指定要處理的輸入數據存放路徑
		FileOutputFormat.setOutputPath(wcjob, new Path("hdfs://weekend101:9000/wc/output3/"));

		// 指定處理結果的輸出數據存放路徑
		wcjob.waitForCompletion(true);

	}
}