MapReduce是什么？能否解释一下长尾词在其中的作用？

本文共计1018个文字，预计阅读时间需要5分钟。

javaMapReduce程序示例：package com.zhiyou100;

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.LongWritable;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;

public class MapReduceExample {

public static class TokenizerMapper extends Mapper {

private final static IntWritable one=new IntWritable(1); private Text word=new Text();

public void map(Object key, Text value, Context context) throws IOException, InterruptedException { String[] tokens=value.toString().split(\\s+); for (String token : tokens) { word.set(token); context.write(word, one); } } }

public static class IntSumReducer extends Reducer { private IntWritable result=new IntWritable();

public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int sum=0; for (IntWritable val : values) { sum +=val.get(); } result.set(sum); context.write(key, result); } }

public static void main(String[] args) throws Exception { Configuration conf=new Configuration(); Job job=Job.getInstance(conf, word count); job.setJarByClass(MapReduceExample.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); }}

MapReduce

package com.zhiyou100; 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.LongWritable; 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; public class WordCount { // 定义 map public static class WordCountMap extends Mapper { private String[] infos; private Text oKey = new Text(); private final IntWritable oValue = new IntWritable(1); /* * context 它的 write 方法 可以传递给 reducer * * LongWritable key, Text value 文本文件的每一个 kv 代表这文本文件里面的一行数据 * key： 该行第一个字符在 该文本中的 偏移量 * val：该文本文件的内容 */ @Override protected void map(LongWritable key, Text value, Mapper .Context context) throws IOException, InterruptedException { // 解析一行数据，转换成一个单词组成的数组 infos = value.toString().split("\\s"); for (String i : infos) { // 把单词形成 kv 对发送给 reducer (单词, 1) oKey.set(i); context.write(oKey, oValue); } } } // 定义 reducer // reducer 的输入类型与 map 的输出类型相对应 public static class WordCountReducer extends Reducer { private int sum; private IntWritable oValue = new IntWritable(0); // 聚合 @Override protected void reduce(Text key, Iterable values, Reducer .Context context) throws IOException, InterruptedException { sum = 0; for(IntWritable value : values){ sum += value.get(); } // 输出 kv (单词, 单词的计数) oValue.set(sum); context.write(key, oValue); } } // 组装一个 job（一次数据处理的过程） 到 mr 引擎上执行 public static void main(String[] args) throws Exception { // 构建一个 configuration ，用来配置 hdfs 的位置和 mr 的参数 // 运行mapreduce程序前都要初始化Configuration，该类主要是读取mapreduce系统配置信息，这些信息包括hdfs还有mapreduce，也就是安装hadoop时候的配置文件 Configuration configuration = new Configuration(); // 创建 job 对象 Job job = Job.getInstance(configuration); job.setJarByClass(WordCount.class); job.setJobName("第一个作业：wordcount"); // 配置 mr 执行类 job.setMapperClass(WordCountMap.class); job.setReducerClass(WordCountReducer.class); // 设置 输出 kv 类型 // map 的 输出 kv 类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // reducer -- 整个 mrjob 的最终输出 kv 类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 当 map 的输出 kv 类型和 mrjob 的最终输出 kv 类型一致的时候可以不用配置 MapOutputKey 的类型 // 两者不一致的时候都要设置 // 设置数据源(待处理) // 设置 mrjob 处理的数据文件位置， // path 可以指定一个文件也可以制定一个文件夹 // 指定文件就处理该文件，如果是文件夹就处理该文件夹下所有的子文件 Path inputPath = new Path("/README.TXT"); // 可以通过多次调用该方法给mrjob 设置多个处理文件的路径 FileInputFormat.addInputPath(job, inputPath); // 设置目标数据的存放位置（设置mrjob 的最终输出结果位置） // 这个路径是一个目录，不能是一个文件，而且当前 hdfs 上不能有这个目录 Path outputPath = new Path("/test/output4"); // 如果存在 path 路径，就删除 outputPath.getFileSystem(configuration).delete(outputPath,true); // 一个 mrjob 只能有一个 输出目录 FileOutputFormat.setOutputPath(job, outputPath); // 启动作业，分布式计算提交给 mr 引擎 // 是否打印处理过程中 输出的日志 boolean result = job.waitForCompletion(true); System.exit(result?0:1); } }