1. 从一次MapReduce任务卡死说起为什么必须搞懂MR的工作机制刚接触Hadoop那会儿我遇到过一个至今印象深刻的场景一个看起来逻辑很简单的MapReduce任务输入数据不过几百MB却跑了将近四十分钟还没结束。打开Web UI一看Map阶段早就100%了Reduce阶段却卡在某个百分比一动不动。当时我的第一反应是数据倾斜但排查后发现分区是均匀的第二反应是资源不够可集群明明很空闲。最后把日志翻出来逐行看才发现问题出在Shuffle阶段——大量中间数据在磁盘上反复溢写Reduce端拉取数据时又频繁触发合并整个链条被拖垮了。那次经历让我意识到一件事很多人学MapReduce学的是怎么写Mapper和Reducer但真正决定一个作业跑得快不快、稳不稳的是它背后那套工作机制——数据怎么切分、键值对怎么流动、Shuffle怎么调度、任务之间怎么协调。你写的代码只是冰山露出水面的一角水面下那套运转逻辑才是大头。这篇内容就是围绕MapReduce的工作机制展开的。我会从作业提交开始一路讲到Map任务、Shuffle与Sort、Reduce任务、任务调度与容错把每个环节到底发生了什么讲清楚。适合两类人看一类是刚学完WordCount、想搞清楚框架到底帮我干了什么的初学者另一类是作业能跑但总出问题、想从机制层面找到优化方向的开发者。读完之后你至少能做到两件事一是看到一个MR作业的日志和计数器能大致判断瓶颈在哪二是知道哪些参数在什么场景下该调、为什么调。需要说明的是下面涉及的参数名和默认值都是基于Hadoop 2.x/3.x的常见实践总结不同发行版和版本之间可能有细微差异实际使用时以你集群的官方文档为准。另外文中提到的所有场景和案例都是我在学习和实践中遇到的典型情况不涉及任何具体机构或项目。2. 一个MR作业从提交到运行中间到底经历了什么2.1 客户端提交不是把代码丢过去就完事很多人以为提交作业就是hadoop jar xxx.jar一敲剩下的交给集群。其实客户端在提交阶段做了不少事理解这些事有助于你排查作业还没跑就失败这类问题。当你在客户端调用Job.waitForCompletion(true)时框架会先做几件事检查输入输出路径是否合法比如输出目录不能已存在、计算输入分片InputSplit、把作业所需的jar包、配置文件和分片信息上传到HDFS上的一个临时目录。这个临时目录通常在/tmp/hadoop-yarn/staging/下面作业ID就是它的名字。这里有个容易被忽略的点输入分片的计算是在客户端完成的。分片的大小直接决定了后面会启动多少个Map任务。默认情况下分片大小等于HDFS的块大小比如128MB。如果你有一个1GB的文件块大小128MB那就会算出8个分片对应8个Map任务。但分片大小是可以调的通过mapreduce.input.fileinputformat.split.maxsize和minsize来控制。为什么有人会调小分片因为Map任务数多了并行度上去了但每个任务处理的数据少了调度开销也上去了——这是个权衡。还有一个细节客户端上传jar包时如果作业依赖了第三方库你得用-libjars参数把这些jar也带上否则任务在集群节点上跑的时候会报ClassNotFoundException。我见过不少新手卡在这里本地跑得好好的一上集群就挂。2.2 资源申请ApplicationMaster的诞生作业提交到ResourceManager之后RM会先分配一个容器Container来启动ApplicationMasterAM。AM是整个作业的管家它负责后续所有任务的调度和监控。AM启动后第一件事是向RM注册自己然后开始为Map任务和Reduce任务申请资源。这里有个关键参数mapreduce.job.reduce.slowstart.completedmaps默认值是0.05。意思是当5%的Map任务完成后AM就会开始为Reduce任务申请资源。为什么不等Map全跑完再申请因为Reduce任务启动后需要先跑Shuffle阶段去拉取Map的输出早点启动可以让Shuffle和Map并行缩短整体时间。但如果Map完成得太少就启动ReduceReduce会空等反而浪费资源。这个值设多少合适取决于你的Map任务平均耗时和集群资源紧张程度。AM和RM之间的通信是心跳机制。AM定期向RM汇报进度和资源需求RM则把可用的资源分配给AM。如果AM长时间不心跳RM会认为它挂了重新启动一个AM——这就是所谓的AM重试。2.3 任务分配Container不是随便给的AM拿到RM分配的资源后会根据数据的本地性Data Locality来决定把Map任务分配到哪个节点。本地性分三个级别数据本地性Data Local任务被分配到数据所在的节点不需要网络传输最快。机架本地性Rack Local任务被分配到数据所在机架的另一台节点需要走机架内网络。跨机架Off Switch任务被分配到不同机架的节点需要走跨机架网络最慢。AM会优先满足数据本地性。如果一时没有合适的节点它会等待一小段时间通过mapreduce.job.locality.wait控制默认3秒实在等不到才降级。这个等待时间设长了任务启动慢设短了本地性差网络传输多。实际调优时如果集群网络是瓶颈可以适当调大这个值。Reduce任务没有数据本地性的概念因为它的输入来自所有Map任务的输出分布在各个节点上。所以Reduce任务通常是随机分配的。3. Map任务内部读数据、算数据、写数据的三段式3.1 输入分片与RecordReader数据是怎么变成键值对的Map任务的输入是一个InputSplit但InputSplit本身只是一个逻辑概念它记录的是从哪个文件的哪个偏移量开始读多长。真正把数据读出来并转成键值对的是RecordReader。以最常见的TextInputFormat为例它对应的RecordReader是LineRecordReader。这个Reader的工作方式是从分片的起始位置开始读每次读一行把行的字节偏移量作为keyLongWritable类型把行内容作为valueText类型。这里有个细节如果分片的起始位置不是文件的开头Reader会跳过第一行不完整的部分从下一个换行符之后开始读。这就是为什么有时候你会看到某个Map任务处理的行数比预期少一点——不是bug是设计如此。为什么key是偏移量而不是行号因为偏移量是天然可定位的不需要额外维护行号计数器而且对于大文件来说偏移量是Long类型不会溢出。行号在分布式环境下反而不好维护。如果你用的是自定义InputFormat那RecordReader的逻辑就由你自己决定了。比如你要处理的是二进制格式的日志可能每次读一个固定长度的记录key设计成记录序号或者时间戳。这里的原则是key的设计要服务于后续的排序和分区需求。因为Map输出的key会被排序和分区输入key虽然不直接参与但会影响你Mapper里的处理逻辑。3.2 Mapper的输出环形缓冲区才是核心Mapper的map()方法每处理一个键值对就调用一次context.write(k, v)。这个write操作并不是直接写到磁盘而是先写到一个内存中的环形缓冲区Circular Buffer。这个缓冲区默认大小是100MB由mapreduce.task.io.sort.mb控制。环形缓冲区的设计很巧妙它是一块连续的内存有两个指针——一个记录当前写入位置 equator一个记录已溢写到磁盘的位置。当写入的数据达到缓冲区大小的80%由mapreduce.map.sort.spill.percent控制默认0.8时一个后台线程就会启动把缓冲区里的数据溢写到磁盘。为什么是80%而不是100%因为要留出空间给溢写期间继续写入的数据避免阻塞Mapper。缓冲区里存的是什么不是原始的键值对而是序列化后的键值对加上元数据。元数据包括分区号partition和key的起始位置。分区号决定了这条数据最终会被哪个Reduce任务处理默认是用key的hashCode对Reduce任务数取模。元数据的存在是为了在溢写时能快速定位每条记录而不需要反序列化。这里有个性能相关的点如果你的key是自定义类型一定要确保hashCode()方法实现得合理。如果所有key的hashCode都一样那所有数据都会分到同一个分区Reduce端就会严重倾斜。我见过有人用自定义对象做key但忘了重写hashCode结果所有数据挤到一个Reduce里跑了几个小时。3.3 溢写与合并磁盘上的排序是怎么做的当缓冲区触发溢写时后台线程会做三件事第一按分区号排序。缓冲区里的数据先按partition排序这样同一个分区的数据会连续存放。第二在每个分区内按key排序。如果定义了Comparator就用自定义的比较器否则用key的默认比较器。第三如果配置了Combiner在排序后、写磁盘前调用Combiner。Combiner的作用是在Map端先做一次局部聚合减少写到磁盘和传输到Reduce端的数据量。Combiner不是万能的。它只适用于满足结合律和交换律的操作比如求和、求最大值。求平均值就不能直接用Combiner因为avg(a, b, c)不等于avg(avg(a,b), c)。如果你硬要用结果会错。我见过有人用Combiner做平均值最后结果偏小排查了半天才发现是Combiner的问题。每次溢写会生成一个spill文件。当Mapper处理完所有输入后可能会产生多个spill文件。在任务结束前这些spill文件会被合并成一个最终的输出文件。合并时也是按分区、按key排序的所以最终输出文件是整体有序的。这个合并过程由mapreduce.task.io.sort.factor控制默认一次合并10个文件。如果spill文件很多可能需要多轮合并。最终输出的文件放在Map任务所在节点的本地磁盘上不是HDFS。为什么因为这是中间数据Reduce任务会来拉取不需要持久化到HDFS。而且本地磁盘读写比HDFS快能减少Shuffle时间。但这也带来一个问题如果Map任务所在的节点挂了中间数据就丢了整个作业得重跑。这是MapReduce容错机制的一部分后面会讲。4. Shuffle与Sort整个MR里最重的环节4.1 Reduce端的拉取Fetcher是怎么工作的Map任务完成后Reduce任务需要去每个Map任务所在的节点拉取属于自己的那部分数据。这个过程叫Shuffle。Reduce任务会启动多个Fetcher线程由mapreduce.reduce.shuffle.parallelcopies控制默认5个并行拉取数据。每个Fetcher的工作流程是先向Map任务所在的NodeManager发HTTP请求获取Map输出文件的位置然后从对应位置拉取数据。拉取过来的数据先放在内存缓冲区里如果内存不够就写到磁盘。这里有个参数mapreduce.reduce.shuffle.input.buffer.percent默认0.7表示Reduce任务堆内存的70%可以用于存储拉取过来的数据。为什么Shuffle这么容易成为瓶颈因为它涉及大量的网络传输和磁盘IO。如果Map输出数据量很大网络带宽就会被吃满如果Reduce端内存不够频繁溢写磁盘磁盘IO也会成为瓶颈。我前面提到的那个卡死的作业就是因为在Shuffle阶段反复溢写磁盘IO打满了。优化Shuffle的一个常见手段是在Map端做Combiner减少传输的数据量。另一个手段是调整Reduce端的拉取并行度如果网络带宽充足可以适当调大parallelcopies。还有一个手段是使用压缩把Map输出压缩后再传输减少网络传输量。压缩算法可以选择Snappy或LZ4它们压缩解压速度快适合中间数据。4.2 归并排序拉取来的数据怎么变成有序的Reduce端拉取到的数据是来自多个Map任务的、已经按key排好序的片段。Reduce任务需要把这些片段合并成一个整体有序的序列这个过程叫归并排序Merge Sort。归并排序是分轮进行的。每一轮从待合并的片段中选出若干个由mapreduce.task.io.sort.factor控制默认10合并成一个更大的有序片段。如果片段数量很多可能需要多轮合并。合并过程中如果内存不够也会溢写到磁盘。这里有个细节Reduce端的排序和Map端的排序用的是同一个Comparator。如果你在Job里设置了自定义的Comparator那Map端和Reduce端都会用它。这保证了排序的一致性。归并排序完成后Reduce任务就得到了一个按key整体有序的输入流。接下来就是调用Reducer的reduce()方法每次处理一个key及其对应的所有value。4.3 分组为什么Reducer一次收到一个key的所有valueReducer的reduce()方法签名是reduce(K key, IterableV values, Context context)。注意第二个参数是IterableV不是单个value。这意味着框架会把相同key的所有value聚在一起一次性传给Reducer。这个聚在一起的过程叫分组Grouping。分组是基于排序结果做的因为数据已经按key排好序了所以相同key的记录是连续的框架只需要顺序扫描遇到key变化就开启新的一组。分组用的比较器是GroupingComparator。默认情况下它和排序用的Comparator是同一个。但你可以单独设置GroupingComparator实现按key排序但按另一套规则分组的效果。比如你有复合key年份月份排序时按年份月份排但分组时只按年份分这样Reducer一次就能拿到某一年的所有月份数据。这个技巧在做二次排序时非常有用。我见过一个典型场景统计每个用户每天的访问次数但需要按用户分组、按时间排序。这时候就可以用复合key用户ID时间戳排序用完整key分组只用用户ID。这样Reducer一次拿到一个用户的所有记录而且记录是按时间排好序的处理起来很方便。5. Reduce任务与输出最后一步也有讲究5.1 Reducer的逻辑你的代码在这里被执行Reducer的reduce()方法被调用时框架已经完成了Shuffle和Sort数据是按key分组、组内按value有序的如果value也实现了排序的话。你在reduce()里写的逻辑就是对这个key及其所有value做处理。这里有个常见的误解很多人以为Reducer是逐个value处理的其实框架是把一个key的所有value都准备好之后才调用一次reduce()。所以如果你的value集合很大可能会撑爆内存。比如某个key对应了几百万个valueIterableV遍历时如果全部加载到内存就会OOM。实际上框架是流式读取的不会一次性加载所有value但如果你在reduce()里把value都塞进一个List那就真的会OOM。Reducer的输出通过context.write(k, v)写出最终写到HDFS上。输出的格式由OutputFormat决定默认是TextOutputFormat每条记录一行key和value用制表符分隔。5.2 输出提交为什么输出目录不能预先存在Hadoop有个规定作业的输出目录必须不存在否则提交时会报FileAlreadyExistsException。这个设计是为了防止误覆盖已有数据。如果你确实想覆盖得先手动删除或者用FileOutputFormat.setOutputPath之前先检查。为什么这么设计因为MapReduce作业的输出是分布式写入的多个Reduce任务可能同时往同一个目录写文件。如果目录已存在且有数据新写入的数据和旧数据混在一起很难区分哪些是本次作业产生的。所以框架干脆要求输出目录必须是新的。Reduce任务写输出时每个Reduce任务会写一个文件文件名通常是part-r-00000、part-r-00001这样的格式。如果你有多个Reduce任务最终输出就是多个文件。这也是为什么有时候你会看到输出目录里有一堆小文件——Reduce任务数太多了。5.3 推测执行慢任务的处理方式如果一个Map或Reduce任务跑得特别慢比如某个节点的磁盘有问题它会拖慢整个作业。Hadoop的应对机制是推测执行Speculative ExecutionAM会为慢任务启动一个备份任务哪个先完成就用哪个的结果另一个被kill掉。推测执行默认是开启的由mapreduce.map.speculative和mapreduce.reduce.speculative控制。但在某些场景下推测执行反而有害。比如你的任务有副作用往外部系统写数据两个备份任务同时写就会重复写。这时候应该关掉推测执行。还有一种情况如果集群资源紧张推测执行会占用额外资源导致其他任务更慢。所以是否开启推测执行要根据具体场景权衡。6. 任务调度与容错作业跑挂了怎么办6.1 任务失败的处理重试不是无限的Map或Reduce任务失败时AM会重新调度这个任务。默认情况下Map任务最多重试4次Reduce任务最多重试4次由mapreduce.map.maxattempts和mapreduce.reduce.maxattempts控制。如果超过重试次数还是失败整个作业就失败了。为什么要有重试次数限制因为如果某个任务因为代码bug一直失败无限重试只会浪费资源。重试机制主要应对的是偶发故障比如节点临时不可用、网络抖动。如果某个节点上失败的任务特别多AM会把这个节点加入黑名单不再往它上面分配任务。这个机制叫黑名单机制。黑名单有有效期过一段时间会自动解除避免节点永久被拉黑。6.2 ApplicationMaster失败整个作业的重启AM本身也可能失败。如果AM挂了RM会检测到通过心跳超时然后启动一个新的AM来接管作业。新AM会从之前的状态恢复重新调度未完成的任务。这个过程对用户是透明的但作业会多花一些时间。AM的重试次数由yarn.app.mapreduce.am.max-attempts控制默认2次。如果AM连续失败超过这个次数作业就彻底失败了。6.3 数据本地性与容错的权衡前面提到Map任务优先分配到数据所在的节点。但如果那个节点挂了数据还在HDFS上因为有副本AM会把任务分配到其他有副本的节点。这就是HDFS副本机制和MapReduce容错机制的配合。如果所有副本所在的节点都挂了概率极低那任务就只能从其他节点读数据走网络传输。这种情况下作业会变慢但不会失败。7. 从计数器看机制怎么用计数器定位瓶颈7.1 计数器是什么框架埋好的监控点MapReduce框架内置了很多计数器用来统计作业运行过程中的各种指标。比如MAP_INPUT_RECORDS记录Map任务读了多少条记录MAP_OUTPUT_RECORDS记录Map输出了多少条REDUCE_INPUT_RECORDS记录Reduce读了多少条REDUCE_OUTPUT_RECORDS记录Reduce输出了多少条。这些计数器不是随便设的它们对应着工作机制里的关键节点。比如MAP_OUTPUT_RECORDS和REDUCE_INPUT_RECORDS应该相等如果没丢数据如果不等说明Shuffle阶段有问题。MAP_OUTPUT_BYTES和REDUCE_SHUFFLE_BYTES可以帮你判断Shuffle的数据量有多大。7.2 用计数器判断数据倾斜数据倾斜是MR作业最常见的性能问题之一。判断方法很简单看各个Reduce任务的REDUCE_INPUT_RECORDS。如果某个Reduce任务的记录数远大于其他任务那就是倾斜了。倾斜的原因通常是key分布不均匀。比如按用户ID分区但某个大V用户的数据量是普通用户的几百倍。解决办法有几种一是自定义Partitioner把大key打散二是在key上加随机前缀让数据均匀分布到多个Reduce然后再做二次聚合。7.3 用计数器判断Shuffle瓶颈如果REDUCE_SHUFFLE_BYTES很大而REDUCE_INPUT_RECORDS相对较小说明数据压缩率不高或者有很多冗余数据。这时候可以考虑在Map端加Combiner或者启用Map输出压缩。如果SPILLED_RECORDS溢写记录数很大说明内存缓冲区不够频繁溢写磁盘。可以调大mapreduce.task.io.sort.mb或者调小mapreduce.map.sort.spill.percent让溢写更早触发但每次溢写的数据量更少。8. 几个容易踩的坑和实操建议8.1 坑一Combiner用错导致结果错误前面提过Combiner只适用于满足结合律和交换律的操作。但很多人不看这个前提直接拿Reducer的逻辑当Combiner用。比如做去重统计Reducer是Set.addCombiner也用Set.add看起来没问题但如果Combiner的输出类型和Reducer的输入类型不一致就会报错。更隐蔽的是平均值场景。avg(1, 2, 3) 2但avg(avg(1,2), 3) avg(1.5, 3) 2.25结果错了。这种错误不会报异常只会让结果偏掉排查起来很费劲。建议用Combiner之前先问自己三个问题——操作满足结合律吗满足交换律吗输入输出类型一致吗三个都是是才能用。8.2 坑二Reduce任务数设成1导致性能瓶颈Reduce任务数由mapreduce.job.reduces控制默认是1。很多人不设这个参数结果所有数据都挤到一个Reduce里Shuffle和Reduce阶段都成了单点瓶颈。Reduce任务数设多少合适一般建议是0.95乘以集群的Reduce槽位数或者根据数据量估算。如果Map输出数据量是10GB每个Reduce处理1GB左右比较合适那就设10个Reduce。但也不是越多越好Reduce任务太多会产生很多小文件而且调度开销也大。8.3 坑三忽略了JVM重用默认情况下每个Map或Reduce任务都会启动一个新的JVM。如果任务很多JVM启动和关闭的开销会很大。可以通过mapreduce.job.jvm.numtasks开启JVM重用让一个JVM跑多个任务。这个值设成-1表示无限重用但实际中设成10到20比较合适避免内存泄漏累积。8.4 实操建议调优从哪下手如果你拿到一个跑得慢的MR作业建议按这个顺序排查先看计数器确认有没有数据倾斜各Reduce的输入记录数是否均匀。再看Shuffle数据量判断是否需要加Combiner或压缩。然后看溢写次数判断内存缓冲区是否够用。最后看任务数判断Map和Reduce的并行度是否合理。调优不是一蹴而就的每次改一个参数跑一遍看效果记录对比数据。我自己的习惯是维护一个调优记录表把每次改的参数、改之前之后的运行时间、计数器变化都记下来时间长了就能形成直觉。9. 写在最后机制懂了调优才有方向回过头看MapReduce的工作机制其实不复杂核心就是分而治之四个字把大任务拆成小任务分片把小任务的结果合并ShuffleReduce。但每个环节都有很多细节这些细节决定了作业的性能和稳定性。我自己的体会是刚学的时候觉得这些机制很琐碎不如直接写代码来得实在。但踩过几次坑之后才发现那些琐碎的机制恰恰是解决问题的钥匙。你知道环形缓冲区默认100MB就知道为什么调大io.sort.mb能减少溢写你知道Shuffle是网络密集型的就知道为什么加Combiner和压缩有效你知道推测执行会启动备份任务就知道为什么有副作用的作业要关掉它。机制不是用来背的是用来推理的。遇到问题的时候从机制出发去分析比盲目试参数要高效得多。希望这篇内容能帮你建立起这套推理的框架下次再遇到MR作业出问题你能知道从哪下手、为什么这么下手。