WordCount案例
需求1：統計一堆文件中單詞出現的個數（WordCount案例）
0）需求：在一堆給定的文本文件中統計輸出每一個單詞出現的總次數
1）數據准備：Hello.txt
hello world
dog fish
hadoop
spark
hello world
dog fish
hadoop
spark
hello world
dog fish
hadoop
spark
2）分析
按照mapreduce編程規范，分別編寫Mapper，Reducer，Driver。
3）編寫程序
（1）定義一個mapper類
package com.xyg.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:默認情況下，是mr框架所讀到的一行文本的起始偏移量，Long; * 在hadoop中有自己的更精簡的序列化接口，所以不直接用Long，而是用LongWritable * VALUEIN:默認情況下，是mr框架所讀到的一行文本內容，String;此處用Text * KEYOUT:是用戶自定義邏輯處理完成之后輸出數據中的key,在此處是單詞，String；此處用Text * VALUEOUT，是用戶自定義邏輯處理完成之后輸出數據中的value，在此處是單詞次數，Integer，此處用IntWritable * @author Administrator */ public class WordcountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{ /** * map階段的業務邏輯就寫在自定義的map()方法中 * maptask會對每一行輸入數據調用一次我們自定義的map（）方法 */ @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 1 將maptask傳給我們的文本內容先轉換成String String line = value.toString(); // 2 根據空格將這一行切分成單詞 String[] words = line.split(" "); // 3 將單詞輸出為<單詞，1> for(String word:words){ // 將單詞作為key，將次數1作為value,以便於后續的數據分發，可以根據單詞分發，以便於相同單詞會到相同的reducetask中 context.write(new Text(word), new IntWritable(1)); } } }
（2）定義一個reducer類
package com.xyg.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.Reducer; /** * KEYIN , VALUEIN 對應mapper輸出的KEYOUT, VALUEOUT類型 * KEYOUT，VALUEOUT 對應自定義reduce邏輯處理結果的輸出數據類型 KEYOUT是單詞 VALUEOUT是總次數 */ public class WordcountReducer extends Reducer<Text, IntWritable, Text, IntWritable> { /** * key，是一組相同單詞kv對的key */ @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int count = 0; // 1 匯總各個key的個數 for(IntWritable value:values){ count +=value.get(); } // 2輸出該key的總次數 context.write(key, new IntWritable(count)); } }
（3）定義一個主類，用來描述job並提交job
package com.xyg.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; /** * 相當於一個yarn集群的客戶端， * 需要在此封裝我們的mr程序相關運行參數，指定jar包 * 最后提交給yarn * @author Administrator */ public class WordcountDriver { public static void main(String[] args) throws Exception { // 1 獲取配置信息，或者job對象實例 Configuration configuration = new Configuration(); // 8 配置提交到yarn上運行,windows和Linux變量不一致 // configuration.set("mapreduce.framework.name", "yarn"); // configuration.set("yarn.resourcemanager.hostname", "node22"); Job job = Job.getInstance(configuration); // 6 指定本程序的jar包所在的本地路徑 // job.setJar("/home/admin/wc.jar"); job.setJarByClass(WordcountDriver.class); // 2 指定本業務job要使用的mapper/Reducer業務類 job.setMapperClass(WordcountMapper.class); job.setReducerClass(WordcountReducer.class); // 3 指定mapper輸出數據的kv類型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 4 指定最終輸出的數據的kv類型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 5 指定job的輸入原始文件所在目錄 FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 7 將job中配置的相關參數，以及job所用的java類所在的jar包， 提交給yarn去運行 // job.submit(); boolean result = job.waitForCompletion(true); System.exit(result?0:1); } }
4）集群上測試
（1）將程序打成jar包，然后拷貝到hadoop集群中。
（2）啟動hadoop集群
（3）執行wordcount程序
[admin@node21 module]$ hadoop jar wc.jar com.xyg.wordcount.WordcountDriver /user/admin/input /user/admin/output
5）本地測試
（1）在windows環境上配置HADOOP_HOME環境變量。
（2）在eclipse上運行程序
（3）注意：如果eclipse打印不出日志，在控制台上只顯示
1.log4j:WARN No appenders could be found for logger (org.apache.hadoop.util.Shell).
2.log4j:WARN Please initialize the log4j system properly. 3.log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
需要在項目的src目錄下，新建一個文件，命名為“log4j.properties”，在文件中填入
log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n log4j.appender.logfile=org.apache.log4j.FileAppender log4j.appender.logfile.File=target/spring.log log4j.appender.logfile.layout=org.apache.log4j.PatternLayout log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
需求2：把單詞按照ASCII碼奇偶分區（Partitioner）
0）分析
1）自定義分區
package com.xyg.mapreduce.wordcount; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner; public class WordCountPartitioner extends Partitioner<Text, IntWritable>{ @Override public int getPartition(Text key, IntWritable value, int numPartitions) { // 1 獲取單詞key String firWord = key.toString().substring(0, 1); char[] charArray = firWord.toCharArray(); int result = charArray[0]; // int result = key.toString().charAt(0); // 2 根據奇數偶數分區 if (result % 2 == 0) { return 0; }else { return 1; } } }
2）在驅動中配置加載分區，設置reducetask個數
job.setPartitionerClass(WordCountPartitioner.class);
job.setNumReduceTasks(2);
需求3：對每一個maptask的輸出局部匯總（Combiner）
0）需求：統計過程中對每一個maptask的輸出進行局部匯總，以減小網絡傳輸量即采用Combiner功能。
1）數據准備：hello,txt
方案一
1）增加一個WordcountCombiner類繼承Reducer
package com.xyg.mr.combiner; import java.io.IOException; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class WordcountCombiner extends Reducer<Text, IntWritable, Text, IntWritable>{ @Override protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int count = 0; for(IntWritable v :values){ count += v.get(); } context.write(key, new IntWritable(count)); } }
2）在WordcountDriver驅動類中指定combiner
//9 指定需要使用combiner，以及用哪個類作為combiner的邏輯 job.setCombinerClass(WordcountCombiner.class);
方案二
1）將WordcountReducer作為combiner在WordcountDriver驅動類中指定
//9 指定需要使用combiner，以及用哪個類作為combiner的邏輯 job.setCombinerClass(WordcountReducer.class);
運行程序
需求4：大量小文件的切片優化（CombineTextInputFormat）
0）需求：將輸入的大量小文件合並成一個切片統一處理。
1）輸入數據：准備5個小文件
2）實現過程
（1）不做任何處理，運行需求1中的wordcount程序，觀察切片個數為5
（2）在WordcountDriver中增加如下代碼，運行程序，並觀察運行的切片個數為1
// 如果不設置InputFormat，它默認用的是TextInputFormat.class
job.setInputFormatClass(CombineTextInputFormat.class);
CombineTextInputFormat.setMaxInputSplitSize(job, 4194304);// 4m
CombineTextInputFormat.setMinInputSplitSize(job, 2097152);// 2m
