存储问题解决了，计算问题是如何解决的？
试想一下，要计算一个大小为100G的文件中某个字符出现的次数，应该怎么做？
用一个计算节点读取分布在 HDFS 中的一个个数据块进行计算？那这个计算节点需要多大的内存？即便有这么大的内存，多久能计算完？
如果你是这样的想法，请跳出单机思维。
看下 MapReduce 是怎么做的。
MapReduce 作为大规模计算框架，它的核心思想是这样的：既然一个大文件可以作为多个数据块存储在 HDFS 集群中，那何不将计算的程序直接传输到这些数据块所在的节点进行分布式计算？
以128M（HDFS 默认分割大小）为一个数据块，100G得有800个数据块。如果按照单机思维，最少要进行800次128M的传输。但如果把一个1M大小的程序传输800次，是不是比前者划算？这也是大数据计算中非常重要的一个思想：移动计算比移动数据更划算。
而之所以叫 MapReduce，是因为 MapReduce 将计算分为了 Map 和 Reduce 两个阶段。开发人员在编码时只需要编写 Mapper 和 Reducer 的实现即可，不用关注程序的移动、计算结果的聚合等分布式编程工作。
以统计字符出现次数的代码为例：
public class WordCount {
public static class TokenizerMapper
extends Mapper<Object, Text, Text, IntWritable>{
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 {
StringTokenizer itr = new StringTokenizer(value.toString());
while (itr.hasMoreTokens()) {
word.set(itr.nextToken());
context.write(word, one);
}
}
}
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);
}
}
public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
if (otherArgs.length < 2) {
System.err.println("Usage: wordcount <in> [<in>...] <out>");
System.exit(2);
}
Job job = Job.getInstance(conf, "word count");
job.setJarByClass(WordCount.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
for (int i = 0; i < otherArgs.length - 1; ++i) {
FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
}
FileOutputFormat.setOutputPath(job,
new Path(otherArgs[otherArgs.length - 1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
main方法执行后，最终会将 Mapper 和 Reducer 的实现作为计算任务分发到数据所在的节点。这样，每个计算任务只用计算128M的数据，800个计算任务同时计算就达到了并行计算的效果，从而实现海量数据的快速计算。
并行计算大概是这样：
每个节点都会先执行 Map 任务（TokenizerMapper） ，将字符出现的次数设置为1，并输出为map（key，value）格式。然后执行 Reduce 任务 （IntSumReducer）将相同字符(key)的次数相加，最后将各节点的结果聚合。
总之，MapReduce 解决了海量数据计算的问题，提供 Map 和 Reduce 这样简单的编程模型，也简化了开发人员对大数据计算的编程难度。
MapReduce 是如何进行任务分发的、计算结果是如何聚合的？如果对这些实现细节感兴趣，请关注我，欢迎大家一起交流。