MapReduce之自定義WordCount案例
阿新 • • 發佈:2018-12-16
在一堆給定的文字檔案中統計輸出每一個單詞出現的總次數。
1.分析
mapper階段:
- 將mapstack 傳給我們的文字資訊內容先轉換成string。
- 根據空格將一行切分成單詞。
- 將單詞輸出為<單詞,1>的格式。
reducer階段
- 彙總各個key的個數
- 輸出該key的總數
driver階段
- 獲取配置資訊
- 指定本程式的jar所在的本地路徑
- 關聯mapper和reducer類
- 指定map的輸出資料kv型別
- 指定最終輸出的資料的kv型別
- 指定job的輸入原始位置和輸出位置
- 提交
2. 編寫程式碼
- 定義一個mapper類
package com.atguigu.mapreduce.wordcount; 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; /** * KEYIN:輸入資料的key 檔案的行號 * VALUEIN 每行的輸入資料 * * KEYOUT : 輸出資料的key * VALUEOUT: 輸出資料的value型別 * @author Administrator * */ public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{ //hello world //atguigu atguigu @Override protected void map(LongWritable key, Text value,Context context) throws IOException, InterruptedException { //1.獲取這一行資料 String line = value.toString(); //2.獲取每一個單詞 String[] words = line.split(" "); //3.輸出每一個單詞 for (String word : words) { context.write(new Text(word),new IntWritable(1)); } } }
2.定義一個reducer類
package com.atguigu.mapreduce.wordcount; import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{ @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { // 1.統計所有單詞個數 int count = 0; for (IntWritable value : values) { count+= value.get(); } //2.輸出所有單詞數 context.write(key, new IntWritable(count)); } }
- 定義一個driver類
package com.atguigu.mapreduce.wordcount;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
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;
//驅動主程式
public class WordcountDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//1.獲取job物件資訊
Configuration configuration = new Configuration();
Job job = Job.getInstance(configuration);
//2.設定載入jar位置
job.setJarByClass(WordcountDriver.class);
//3.設定mapper和reducer的class類
job.setMapperClass(WordcountMapper.class);
job.setReducerClass(WordcountReducer.class);
//4.設定輸出mapper的資料型別
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
//5.設定最終資料輸出的型別
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
//6.設定輸入資料和輸出資料路徑
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
//7.submit
boolean result = job.waitForCompletion(true);
System.exit(result?0:1);
}
}
3.執行
- 將上述的類打成jar包,拷貝到hadoop叢集中
- 啟動hadoop叢集
- 執行WordCount程式
開啟web端HDFS檢視結果
將 part-r-00000下載下來檢視,滿足需求。