1. MapReduce核心思想为什么说是“分而治之”的大规模计算范式1.1 从一个实际问题出发处理上亿条日志里的关键词统计很多人第一次接触MapReduce都是从“统计单词出现次数”这个经典例子开始的。但真正让我理解MapReduce价值的是一次实际的日志分析需求线上服务每天产生大约3亿条访问日志每条日志里包含用户行为、请求路径、状态码等信息。老板的需求很朴素——把这一天的日志按请求路径聚合统计每个路径的访问量和平均响应时间。如果日志量是几十万条用Shell脚本或者Python单机处理几秒钟就出结果了。但3亿条单机处理非常吃力——内存扛不住磁盘IO也扛不住跑一个统计任务可能要几个小时而且中间一旦机器内存爆掉整个任务就白跑了。这时候就面临一个选择要么花钱买更大的服务器要么换一种能横向扩展的计算框架。买更大服务器这件事有明显的“天花板”不仅是成本问题单台机器的CPU核数、内存插槽、磁盘通道都是有限制的。这也是MapReduce这类分布式计算框架存在的根本原因与其让一台机器累死不如让一百台机器各干一点活。MapReduce的本质就是把这个“各干一点活”的过程规范化、框架化。它不像你用Python手写多线程、手动分片、手动合并结果那样“自由”但它提供了一套严谨的约定你只需要写两个函数——map和reduce——剩下的分布式调度、数据分发、故障重试、结果聚合全部由框架帮你完成。1.2 MapReduce的“分”与“合”Map阶段和Reduce阶段“MapReduce”这个名字本身就揭示了它的全部秘密。Map映射负责“分”Reduce归约负责“合”。先说Map阶段。输入数据被框架自动切分成若干个分片split每个分片由一个Map任务处理。这里注意一个细节分片的大小是可以配置的默认情况下和HDFS的块大小对齐通常是128MB或256MB。每个Map任务读入分片中的数据逐条调用你写的map()函数把输入记录转换成中间输出的Key-Value对。举例来说如果你要统计单词频率一条输入“hello world hello”会被map()函数拆成三个键值对(hello, 1)、(world, 1)、(hello, 1)。如果输入是日志map()函数可以解析出请求路径输出(path, 1)这样的键值对或者输出(path, response_time)以便后续算平均值。再说Reduce阶段。Map阶段产出的中间结果不会直接写回HDFS而是先经过一个极其重要的“洗牌”过程Shuffle将相同Key的键值对自动合并到一起形成一个(Key, [value1, value2, ...])的列表再交给Reduce任务处理。reduce()函数拿到这个列表后做最终的聚合计算。还是拿单词统计举例Reduce端拿到(hello, [1,1,1])把所有1加起来就是hello的出现次数。这个模型的设计巧妙之处在于Map和Reduce之间是松耦合的。Map任务的输出只要符合Key-Value格式Reduce任务就能按Key聚合至于中间数据如何传输、如何排序、如何分组框架层面已经做了标准实现。你不需要关心“哪台机器跑了哪个Map任务”“中间数据存到哪里”这些全部是透明的。1.3 适合什么场景不适合什么场景在深入学习MapReduce之前先搞清楚边界否则容易用错方向。MapReduce的强项是离线批处理。数据量巨大TB级甚至PB级、计算逻辑相对简单、对延迟要求不高分钟级到小时级、需要跑完一个完整的计算任务比如全量数据统计、全量数据清洗、全量索引构建这些场景是MapReduce的主场。MapReduce不适合的场景也很明确实时计算不行——它的设计目标是高吞吐而非低延迟一个任务从启动到结束光任务调度和Shuffle就要花不少时间复杂的多阶段迭代计算不行——比如机器学习里的梯度下降算法要反复迭代同一份数据MapReduce每个阶段都要把中间结果落盘迭代十几轮的话磁盘IO开销会让人崩溃大量Tuple级随机读写也不行——它不适合像传统关系型数据库那样做单条记录的更新、删除和点查。所以如果你现在的场景是“每天凌晨跑一次大规模报表统计”“每月把上亿条交易数据做一次全量校验”MapReduce依然是一个非常可靠、可维护性很强的选择——虽然现在很多新项目会选Spark或Flink但MapReduce所建立的“Map-Shuffle-Reduce”思维模型是你理解后面所有分布式计算框架的基础。把这一步吃透后面学Spark、Flink会轻松一半。2. 深入拆解Map阶段与Reduce阶段的职责边界2.1 Map阶段输入分片、RecordReader与map()函数Map阶段远不是“写一个map方法”这么简单它内部有一套执行流程。首先框架会把输入数据切成输入分片InputSplit。分片是一个逻辑概念它不物理存储数据只是指定了要处理的数据范围和对应的元数据信息。分片大小默认与HDFS块对齐这意味着如果你HDFS块大小是128MB那么一个1GB的文件会被切成大约8个分片由8个Map任务并行处理。接下来每个Map任务内部有一个RecordReader组件负责按行读取分片中的数据并解析成Key-Value对。默认情况下TextInputFormat的RecordReader会把每行的行首偏移量作为Key整行文本作为Value。也就是说虽然你看到的map()函数参数是(Object key, Text value)但这个key实际上就是行号偏移量业务上用不到它。然后map()函数逐条处理这些键值对处理完通过context.write()输出新的Key-Value对。这个输出会被写入一个环形缓冲区缓冲区默认大小100MB当写入量达到阈值比例默认80%时后台线程开始将数据溢写spill到本地磁盘。溢写之前会先做分区和排序——这一步我后面会详细讲因为这是理解Shuffle的关键。还有一点值得留意map()函数的输入和输出Key-Value类型必须实现Writable接口这是Hadoop自定义的序列化接口。为什么不用Java自带的Serializable因为Java序列化太重了会把类名、继承结构等大量元信息也序列化进去而Hadoop的Writable只序列化字段内容体积小、速度快。你在写MapReduce程序时文本用Text、整数用IntWritable、长整数用LongWritable字符串数组就用Text数组的包装这些类型直接对应了Java里的String、Integer、Long。2.2 Reduce阶段数据分组、reduce()函数与多Reduce任务Reduce阶段接收的是Map端输出的中间数据。但这里有个容易误解的点所有相同Key的键值对都会交给同一个Reduce任务处理无论这些键值对来自哪个Map任务。这也就是Shuffle的核心作用——它完成了一次“全局分组”。每个Reduce任务内部会有分组Grouping的步骤把Key相同的记录归为一组。你写的reduce()函数参数是(Text key, IterableIntWritable values)这个Iterable就是一个分组迭代器每次迭代取出的value都属于同一个Key。注意一个细节这个Iterable里的values只能迭代一次如果你需要多次访问这些值比如既要算总数又要算平均值需要遍历两遍就得先把values复制到List里否则第二遍迭代会得到空集合。Reduce任务的数量由你显式设置默认情况下是1。这意味着所有Key都会涌向同一个Reduce任务也就是单点处理——数据量大的时候性能极差。通常建议把Reduce数量设置成和集群可用CPU核数相关比如job.setNumReduceTasks(8)或更大一点。但也不是越大越好Reduce数量太多会导致每个Reduce处理的数据量过少启动和调度的开销占比反而上升任务完成后输出的小文件也会变多。2.3 中间结果的Key-Value设计原则写MapReduce程序时map()输出什么Key、什么Value看似自由发挥实际上有很强的约束和技巧。核心原则是Key必须是“语义上的分组维度”。因为框架会按Key分组所以你想按什么维度聚合就把什么作为Key。比如按请求路径统计流量Key就是路径字符串按用户ID统计消费总额Key就是用户ID。如果把业务维度放在Value里而Key设置得很粗糙Reduce端还是要自己在values里做二次分组失去了框架分组的优势。Value的设计相对自由但有一个常见场景要注意——Reduce端需要不止一个字段时。比如统计每个路径的访问量和平均响应时间map()输出Key为路径Value如果只是一个IntWritable的响应时间值那Reduce端只能算出总花费时间或总次数算不出平均值因为平均值需要同时知道次数和时间总和。这时候有两种做法第一种写一个自定义Writable类型包含count和time两个字段map()输出这个复合对象Reduce端分别累加最后除法得到平均响应时间。第二种用Text类型把count和time拼成一个字符串比如1:128Reduce端再拆分解析。这种方法简单粗暴但可读性差如果字段里本身含分隔符就麻烦。我建议用自定义Writable虽然代码量多一点但类型安全、扩展性好后续加字段不用改String解析逻辑。后面第4节我会给完整的代码。3. Shuffle机制MapReduce最容易被忽视的“隐形枢纽”3.1 Shuffle全流程从Map端输出到Reduce端输入我见过很多人写MapReduce程序调优时总在纠结map和reduce函数本身写得“对不对”结果发现瓶颈根本不在业务代码而在Shuffle——它是整个MapReduce框架中最复杂、最容易出性能问题的环节。Shuffle横跨Map和Reduce两端具体流程是这样的Map端map()函数调用context.write()后输出数据先进入环形缓冲区。这个缓冲区默认100MB当数据量达到缓冲区大小的80%也就是80MB时后台线程开始溢写。溢写时先把数据按Partitioner分区就是把Key映射到哪个Reduce任务每个分区内部再按Key排序。如果配置了Combiner还会在排序后做一次局部合并减少溢写数据量。溢写会产生多个spill文件Map任务全部结束后这些spill文件会被合并成一个最终的输出文件这个过程叫Merge。Reduce端Reduce任务启动后会从Map端拉取属于自己分区的数据。注意Reduce任务不知道数据在哪它需要向ApplicationMaster询问每个Map任务的输出位置然后并行地下载属于本分区的数据。下载的数据先放在内存内存不够就落盘。全部拉完后再进行一次最终的合并排序把相同Key的记录聚在一起然后才轮到reduce()函数执行。这个流程里每一步都涉及磁盘IO和网络传输所以Shuffle的数据量直接决定了作业的运行时间。优化Shuffle的核心思路就是尽可能减少中间数据的体积——这就要用到分区控制、Combiner和压缩了。3.2 分区、排序、分组对应关系一眼理清不少初学者会把分区Partition、排序Sort、分组Grouping混为一谈其实它们解决的是不同层面的问题。分区解决的是“这个Key交给哪个Reduce任务”。默认的分区器是HashPartitioner它计算(key.hashCode() Integer.MAX_VALUE) % numReduceTasks通过哈希取模把Key均匀分布到各个Reduce。这样可以保证相同Key的键值对必然进入同一个Reduce因为同一Key的哈希值相同。如果你想让某些有业务关联的Key进入同一个Reduce比如所有订单量大的用户ID进同一个Reduce做特殊处理可以自定义Partitioner重写getPartition()方法。排序解决的是“同一个分区里的Key按什么顺序排列”。默认按Key的字典序升序。有些Reduce计算依赖有序数据比如数值统计需要从小到大累加、去重需要跳过连续相同值这时候排序就有用了。你可以在map阶段就做部分排序Reduce阶段再全局归并最终保证每个分组内部的Key是有序的。分组解决的是“对Reduce函数来说哪些Key算一组”。默认情况下两个Key完全相等才分成一组。但有时你想让“语义上相关但不相等”的Key分到同一组——比如Key是句子你想让同一句子不同大小写形式归入一组或者月份字段是2024-01和2024-1你想合并。这时候可以自定义GroupingComparator重写比较逻辑让不完全相等的Key在Reduce端被视为同一组。一句话总结分区控制流向排序保证顺序分组决定视图。三者独立又协作是MapReduce里最考验功力的部分。3.3 Combiner的正确使用方式Combiner是很多人容易用错的一个优化配置。它的作用是在Map端先做一次局部合并减少Shuffle传输和Reduce端的数据量。比如WordCount200个Map任务每个都输出(hello, 1)那你传输到Reduce端的数据就有200份。但如果每个Map任务在本地先做一次reduce操作把(hello, 1)合并成(hello, 67)中间数据就只剩一份。但要注意Combiner不是想怎么用就怎么用。Combiner的输入和输出必须是同一个类型而且它的操作必须是幂等可重复的。最典型的问题是“求平均值”——如果map()输出(path, 128)Combiner把两个值合并成(path, 256)这没有问题因为求和是可结合的但如果Combiner算的是平均值两个127合并成一个128那两个127和一个128这三个值的平均应该怎么算只拿到128这个结果的人是算不出来的。所以求平均值不能直接拿Combiner当Reduce用。我常用的方式是如果Reduce做的是求和、求Count、求最大值、求最小值这类满足交换律和结合律的操作Combiner可以直接复用Reducer类。如果是平均值、中位数这类复杂操作要么在Combiner里输出(总和, 次数)这种二元组并在Reducer里相除要么干脆不用Combiner避免引入错误结果。4. WordCount实战第一个MapReduce程序的完整实现4.1 环境准备用什么版本、怎么快速搭一套本地开发环境写MapReduce程序不一定要先搭一个完整的Hadoop集群——对于学习和调试本地模式Local Mode就够了。我的建议是先在本地把逻辑跑通再上集群跑全量数据这样排查问题的时间成本会低很多。具体环境上我目前用的是Hadoop 3.x系列比如3.3.xJDK 8以上Maven做依赖管理。在Maven项目里的pom.xml中引入dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-client/artifactId version3.3.6/version /dependency如果你不习惯用Maven也可以直接把Hadoop安装包里的所有jar包在share/hadoop/common、share/hadoop/mapreduce、share/hadoop/hdfs等目录都加到classpath里。但说实话用Maven管理会清爽很多换版本、传递依赖都省心。本地模式下不需要启动任何守护进程。Hadoop的LocalJobRunner会自动模拟单机、单进程的环境直接跑在JVM里适合快速验证代码逻辑。同时它也支持访问本地的HDFS——如果不显式指定HDFS地址默认会使用本地文件系统的临时目录file:///tmp/hadoop-${user}。所以你在本地跑通之后代码直接打成jar包扔到集群上运行时通过-fs hdfs://namenode:8020这样的参数切换文件系统。4.2 Mapper实现从一行文本到键值对经典的WordCountmap()函数要做的事非常直观每读入一行文本按空格和标点把它切分成单词每个单词输出(单词, 1)。完整代码如下import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class WordCountMapper extends MapperLongWritable, Text, Text, IntWritable { private static final IntWritable ONE new IntWritable(1); private Text word new Text(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // value 是一行文本key 是该行在文件中的字节偏移量 String line value.toString(); // 简单分词按非字母数字字符切分 String[] words line.split([\\s\\p{Punct}]); for (String w : words) { if (w.isEmpty()) { continue; } word.set(w); context.write(word, ONE); } } }这里我用了正则[\s\p{Punct}]来切分可以一次性处理空格、制表符和标点。这个是比较Python风格的写法Java里也支持。如果你要处理的是日志文件map()里就得写针对日志格式的解析逻辑了比如按空格split后取第几个字段作为Key。4.3 Reducer实现聚合统计并输出Reducer这边做的事情更简单把迭代器里的所有value加起来写出去。import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class WordCountReducer extends ReducerText, IntWritable, Text, IntWritable { private IntWritable result new IntWritable(); Override protected void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException { int sum 0; for (IntWritable val : values) { sum val.get(); } result.set(sum); context.write(key, result); } }如果还想按出现次数降序排序输出默认的Reduce输出是按Key字典序升序排列的。如果你要全局排序需要再加一个控制排序的环节。这里先不展开后面讲进阶技巧时会提到。4.4 Driver主类Job的组装与配置有了Mapper和Reducer还需要一个Driver类来组装整个Job并提交运行。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.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class WordCountDriver { public static void main(String[] args) throws Exception { if (args.length ! 2) { System.err.println(Usage: WordCountDriver input path output path); System.exit(-1); } Configuration conf new Configuration(); Job job Job.getInstance(conf, Word Count); job.setJarByClass(WordCountDriver.class); job.setMapperClass(WordCountMapper.class); job.setCombinerClass(WordCountReducer.class); // 复用Reducer做Combiner job.setReducerClass(WordCountReducer.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); } }这里有几个细节job.setJarByClass(WordCountDriver.class)是必须的它告诉Hadoop去哪里找包含所有依赖类的jar包。如果你在终端里用hadoop jar xxx.jar提交框架会使用你指定的jar包这个设置依然会做校验。我在这里设置了job.setCombinerClass(WordCountReducer.class)因为WordCount的Reduce逻辑是求和满足可结合性Combiner直接复用Reducer类合法且高效。FileInputFormat默认会递归读取指定目录下的所有文件如果你配置了递归输出路径必须不存在否则Hadoop会直接报错——这是为了避免误覆盖历史结果。打包时用Maven的shade插件把依赖打成一个fat jar。实际运行命令如下hadoop jar wordcount-1.0.jar WordCountDriver /input /output4.5 运行与结果验证本地模式运行时如果你用IDE直接跑Driver类需要在IDE里配置程序参数input path和output path比如# 在项目根目录下准备输入文件 echo -e hello world\nhello hadoop\nmapreduce is powerful input.txt # 运行本地模式 hadoop jar target/wordcount-1.0.jar WordCountDriver input.txt output_dir跑完之后查看结果hadoop fs -cat output_dir/part-r-00000如果你是在本地文件系统跑结果文件也在本地直接cat output_dir/part-r-00000即可。正常情况下你会看到每个单词一行统计结果按单词字典序排好。如果结果里每个单词出现了两行或者统计值不对大概率是Combiner配置出了问题或者map()输出Value类型和Reducer的期望不一致——这是新手最容易翻车的两个点。5. MapReduce编程进阶让作业跑得更稳更快的经验5.1 合理设置并行度Map数量和Reduce数量怎么定很多初学者以为Map任务越多并行度越高其实Map任务数是有限制的而且并非越多越好。Map任务数由输入文件的大小和分片大小决定。默认情况下一个128MB的分片对应一个Map任务。如果你的输入是100GB理论上会有大约800个Map任务。如果都是几KB的小文件每个小文件也可能产生一个Map任务这就会出现“一个Map任务处理几KB数据、启动开销远大于计算开销”的极端情况。所以合理规划输入数据的存储结构很重要。把大量小文件合并成SequenceFile或Parquet这类大文件是降低Map任务数、提升整体效率的最有效手段之一。另外在Driver类里可以通过FileInputFormat.setMaxInputSplitSize()调大分片大小来减少Map数但要注意分片过大时单Map任务处理时间也会变长需要根据实际数据分布权衡。Reduce数量的设置更讲究。一个通用经验公式是Reduce数 每个Reduce期望处理的数据量 / 集群可用CPU核数比如总共要处理100GB的中间数据集群有200个可用vcore每个Reduce处理1GB数据比较合适那么设50个Reduce是比较合理的。Reduce太少会导致单个Reduce负载过重、任务长时间跑不完Reduce太多会导致调度开销大、每个Reduce拿到的数据碎不成片、输出小文件过多。5.2 数据倾斜你的作业为什么有的Reduce跑了很久有的却早早干完数据倾斜是MapReduce作业运行时间不可控的“头号元凶”。症状很明显同一个作业里某个Reduce任务运行时间是其他Reduce的几倍甚至几十倍或者某个Reduce任务进度条一直卡在100%附近不动。根本原因在于Key的分布极度不均匀。比如统计用户日活时超级大V用户的行为日志特别多映射到同一个Reduce上其他普通用户的Reduce任务几分钟就结束了大V所在的那个Reduce要跑几个小时。针对倾斜的常规处理手段有几种给倾斜的Key加前缀或随机后缀让它“分身”到多个Reduce。比如大V用户ID后面加_0、_1、_2这样它的数据被分散到3个Reduce上最后再单独跑一个二次聚合把结果合并。如果倾斜只集中在少数几个Key可以用采样预先识别出倾斜Key在map()阶段就单独标记处理不让它走默认分区。用Combiner尽量做局部合并也能显著缓解倾斜程度。实际操作中我一般会先看一眼任务日志定位到那个迟迟不结束的Reduce然后打印它处理的Key分布确认是倾斜后再针对性加随机后缀做二次聚合。5.3 小文件问题MapReduce最容易被低估的性能杀手Hadoop的HDFS本身是为大文件设计的块默认128MB但实际业务里经常出现大量几KB、几十KB的小文件比如日志按小时切分后每个小时的文件非常碎或者上游接口每5分钟吐一个临时文件。小文件带来的连锁反应很严重Map任务数量剧增、NameNode内存压力飙升每个文件都要在NameNode里维护元数据、磁盘寻道开销大、调度和启动任务的资源消耗盖过了实际计算。我的经验是在做MapReduce批处理之前先做一层文件合并。可以用Hadoop自带的工具也可以在上游做数据汇聚时把粒度从“分钟级”调整为“小时级”或“天级”。Hadoop 3.x还引入了Storage Policy和Erasure Coding能缓解一部分小文件带来的副本存储开销但治本的方法还是提前控制文件粒度。5.4 中间数据压缩用CPU换网络和磁盘IOShuffle过程的传输量如果降不下来还有一个性价比极高的优化手段——中间结果压缩。在Driver类里加上配置conf.set(mapreduce.map.output.compress, true); conf.set(mapreduce.map.output.compress.codec, org.apache.hadoop.io.compress.SnappyCodec);Snappy压缩算法速度非常快压缩率在中低水平但CPU开销很小对绝大多数场景都是正收益。开启后Map端溢写的spill文件和Reduce端拉取的数据都会变小网络传输压力明显降低。如果输入文件本身是压缩格式比如Gzip、Bzip2、SnappyHadoop会自动识别并解压。但要注意Gzip不支持分片splittable一个Gzip文件只能由一个Map任务处理大文件配Gzip会让你眼睁睁看着并行度变成1。所以如果输入文件较大且需要高并行度建议存储时用Snappy或者LZO这两种格式支持分片。6. 常见问题与排查技巧实录6.1 任务反复失败从日志里怎么最快定位问题MapReduce作业失败时第一件事是看日志——但日志在哪、看哪个日志是有讲究的。如果你提交的是hadoop jar命令任务运行完控制台会打印每个Task的日志链接格式类似于http://nodemanager:8042/logs/userlogs/application_xxx/container_xxx/。你可以打开对应container的标准输出、标准错误和系统日志。如果任务失败通常syslog里含有业务代码抛出的异常堆栈这是排查的第一手信息。最常见的几类错误ClassNotFoundException说明你打包的jar里没有包含依赖的类检查是不是忘了用shade插件打进依赖。FileAlreadyExistsException输出目录已经存在。Hadoop不会自动覆盖需要先删除旧目录。GC overhead limit exceeded或OutOfMemoryError内存设置不合理检查Map和Reduce的堆内存配置。Killed by AM说明Container被资源管理器主动杀掉通常是内存超限或Yarn队列资源配额不够。我个人的排查顺序是先看ApplicationMaster页面上的“失败任务列表”然后点进失败的Task里看syslog再对应异常类型去处理。不建议一上来就翻几十个Task的日志浪费时间。6.2 内存溢出和GC频繁参数怎么调Map任务默认的堆内存上限一般是1GB左右如果你的一个Map任务要处理的行数特别多、value特别长很容易触发OOM。可以通过参数调整hadoop jar myjob.jar \ -Dmapreduce.map.memory.mb2048 \ -Dmapreduce.reduce.memory.mb4096 \ ...mapreduce.map.memory.mb控制的是Yarn分配给Map Container的内存上限建议同时调整对应JVM堆大小mapreduce.map.java.opts比如设为-Xmx1536m。注意Container内存要大于JVM堆内存因为还有内存开销JVM自身、代码缓存、直接内存等。如果你发现GC非常频繁但还没有OOM可以先开启map端的中间输出压缩前面提过的Snappy减少内存和磁盘间的数据流动另一个角度是把环形缓冲区的默认阈值调小减少一次溢写携带的数据量。6.3 结果不符合预期数据去哪儿了写完了MapReduce跑完了结果却对不上这种问题在开发阶段很常见。首先检查一下你的map()输出Key和预期是否一致。有时候你以为自己切分好了单词结果分词正则里有特殊字符没处理干净导致hello.和hello被当成两个不同的Key。用上面WordCount的示例分词正则漏掉句号是很常见的事。其次检查Reduce端是否用到了Iterable的重复遍历。前面提过Iterable里的values只能遍历一次。如果你先循环一遍算总和、再循环一遍算平均值第二次循环会得到空集合最终结果自然是不对的。解决办法是把values复制到List里再做多轮计算。最后注意如果你的Job设置了多个Reduce任务输出文件名是多份比如part-r-00000、part-r-00001。你聚合所有part文件才是完整结果少了哪一份都可能导致统计值不对。6.4 如何调试MapReduce程序本地跑、单元测试、小数据量验证我强烈建议开发时遵循“先小后大”的原则先在本地用几MB的样例数据跑通再上集群用几十GB的数据做验证最后才跑全量。本地调试时最直接的方式是在IDE里直接运行Driver类。它能实时看到日志和异常堆栈打断点调试也很方便。唯一问题是本地模式默认单进程运行效率不高但调试逻辑足够了。如果你需要更细粒度的单元测试可以用Hadoop自带的测试框架如MRUnit对Mapper和Reducer分别写测试用例。这样可以在不启动任何Hadoop环境的情况下快速验证map和reduce的输入输出逻辑。项目里引入dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-minicluster/artifactId version3.3.6/version scopetest/scope /dependencyMiniCluster是Hadoop提供的单机模拟集群会启动一个完整的HDFS和YARN环境非常适合做集成测试。最后再分享一个小技巧如果你排查问题一直无头绪给map()和reduce()函数里临时加上业务日志比如把每个Key的value打印前十条用System.err.println输出到syslog里。跑完任务直接去container日志里看。这个方法虽然朴素但很多诡异的“数据对不上”问题就是这样被定位出来的——尤其是当你不确定上游数据到底长什么样时打印原始输入往往比看中间代码更有效。