MapReduce实现单词计数：WordCount
单词计数的文本信息(hello.txt)：
hello can i help you
i have a dream
maybe you can help me
实现过程：
Map过程：并行读取文本，对读取的单词进行Map操作，每个词将会形成。
第一行将会形成：
第二行生成
第三行生成
Map会根据key值，自动按照ASCII值进行排序，所以，相同key值得K/V值，将会被放到一起
Reduce过程：对map的结果进行合并，计算，
Map将key相同得值放到了一起，所以当Reduce拿到从Map传过来得数据，可以理解为如下
而Reduce进行最后得合并，就会得到如下得结果
最后输出
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-g8x0hrTQ-1578386385884)(C:\Users\Lenovo\AppData\Roaming\Typora\typora-user-images\image-20200107162331181.png)]
Java Api 编写
public class WordCountApp {
//静态类，用于继承Mapper类，用于继承Mapper方法
public static class MyMapper extends Mapper{//mapper的四种参数类型(源码使用了泛型，具体可以查看源码)
//IntWritable与Text是Hadoop中的两个参数，可以简单的理解为是
//在hadoop中的int与text格式
private final static IntWritable one =new IntWritable(1);
private Text word =new Text();
//map操作
public void map(Object key ,Text value ,Context context)throws IOException,InterruptedException{
//StringTokenizer是Java.util中的类，用于将字符串切割成
//一个字符串数组(用" "把各个字符分隔开)。
StringTokenizer itr =new StringTokenizer(value.toString());
//遍历文本中一行的所有字符串
while (itr.hasMoreTokens()){
word.set(itr.nextToken());
//写入到map的输出文件中，这个context用于Reduce的接收
context.write(word,one);
}
}
}
//静态类，用于继承Reduce类，编写Reduce方法
public static class MyReduce extends Reducer{
//统计计数，即记录单词的出现次数，最后输出的辅助类
private final static IntWritable result =new IntWritable();
//reduce操作
public void reduce(Text key , Iterablevalues, Context context)throws IOException,InterruptedException{
//记录单词的出现个数，因为在Reduce中没有int类型，所以要
//用IntWritable进行包装
int sum=0;
//上面讲到，Reduce拿到Map的值的形式，这个Reduce会执行在每
//一个块中，用于计数，所以每一个单词都是从0开始计数
for (IntWritable val : values){
sum +=val.get();
}
//int sum包装进 IntWritable
result.set(sum);
//写入到最终会输出的地方
context.write(key,result);
}
}
//主方法
public static void main(String[] args) throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {
//本此编码是在Windowx编程，所以要加Hdfs的路径
String INPUT_PATH="hdfs://10.30.60.3:8020/a";
String OUTPUT_PATH="hdfs://10.30.60.3:8020/b";
Configuration configuration =new Configuration();
final FileSystem fileSystem=FileSystem.get(new URI(INPUT_PATH),configuration);
//确保INPUT的路径存在，而OutPut得路径会在最后生成结果时生成、
//所以确保OUTPUT得文件不存在
if(fileSystem.exists(new Path(OUTPUT_PATH))){
fileSystem.delete(new Path(OUTPUT_PATH),true);
}
//设置job
Job job =Job.getInstance(configuration,"WordCountApp");
//设置主类
job.setJarByClass(WordCountApp.class);
//设置mapper
job.setMapperClass(MyMapper.class);
//设置Mapper的输出格式
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
//设置Reduce
job.setReducerClass(MyReduce.class);
//设置Reduce格式
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
//设置输入格式
job.setInputFormatClass(TextInputFormat.class);
Path inputPath=new Path(INPUT_PATH);
FileInputFormat.addInputPath(job,inputPath);
//设置输出格式
job.setOutputValueClass(TextOutputFormat.class);
Path outputPath=new Path(OUTPUT_PATH);
FileOutputFormat.setOutputPath(job,outputPath);
//提交
System.exit(job.waitForCompletion(true)? 0:1);
}
}
)
