package cn.hdfs.mr.example; import java.io.IOException; import java.util.Random; import java.util.StringTokenizer; 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.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import cn.hdfs.utils.ConfigUtils; /** * * @author dengjie * @date 2015年03月13日 * @description Wordcount的例子是一个比较经典的mapreduce例子，可以叫做Hadoop版的hello world。 * 它将文件中的单词分割取出，然后shuffle，sort（map过程），接着进入到汇总统计 * （reduce过程），最后写道hdfs中。基本流程就是这样。 */ public class WordCount { private static Logger log = LoggerFactory.getLogger(WordCount.class); public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> { private final static IntWritable one = new IntWritable(1); private Text word = new Text(); /* * 源文件：a b b * * map之后： * * a 1 * * b 1 * * b 1 */ public void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString());// 整行读取 while (itr.hasMoreTokens()) { word.set(itr.nextToken());// 按空格分割单词 context.write(word, one);// 每次统计出来的单词+1 } } } /* * reduce之前： * * a 1 * * b 1 * * b 1 * * reduce之后: * * a 1 * * b 2 */ public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } @SuppressWarnings("deprecation") public static void main(String[] args) throws Exception { Configuration conf1 = new Configuration(); Configuration conf2 = new Configuration(); long random1 = new Random().nextLong();// 重定下输出目录1 long random2 = new Random().nextLong();// 重定下输出目录2 log.info("random1 -> " + random1 + ",random2 -> " + random2); Job job1 = new Job(conf1, "word count1"); job1.setJarByClass(WordCount.class); job1.setMapperClass(TokenizerMapper.class);// 指定Map计算的类 job1.setCombinerClass(IntSumReducer.class);// 合并的类 job1.setReducerClass(IntSumReducer.class);// Reduce的类 job1.setOutputKeyClass(Text.class);// 输出Key类型 job1.setOutputValueClass(IntWritable.class);// 输出值类型 Job job2 = new Job(conf2, "word count2"); job2.setJarByClass(WordCount.class); job2.setMapperClass(TokenizerMapper.class); job2.setCombinerClass(IntSumReducer.class); job2.setReducerClass(IntSumReducer.class); job2.setOutputKeyClass(Text.class); job2.setOutputValueClass(IntWritable.class); // FileInputFormat.addInputPath(job, new // Path(String.format(ConfigUtils.HDFS.WORDCOUNT_IN, "test.txt"))); // 指定输入路径 FileInputFormat.addInputPath(job1, new Path(String.format(ConfigUtils.HDFS.WORDCOUNT_IN, "word"))); // 指定输出路径 FileOutputFormat.setOutputPath(job1, new Path(String.format(ConfigUtils.HDFS.WORDCOUNT_OUT, random1))); FileInputFormat.addInputPath(job2, new Path(String.format(ConfigUtils.HDFS.WORDCOUNT_IN, "word"))); FileOutputFormat.setOutputPath(job2, new Path(String.format(ConfigUtils.HDFS.WORDCOUNT_OUT, random2))); boolean flag1 = job1.waitForCompletion(true);// 执行完MR任务后退出应用 boolean flag2 = job1.waitForCompletion(true); if (flag1 && flag2) { System.exit(0); } else { System.exit(1); } } }

4.总结

这篇文章就和大家分享到这里，如果在研究的过程有什么问题，可以加群讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！