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); 
} 
} 
}