凌晨两点一个跑了一整晚的离线作业卡在 reduce 67% 不动日志里每隔十分钟刷一次拉取失败这种场面我在做数据平台那几年遇过不止一次。每次往上追最后都绕回同一个东西——MapReduce 的执行模型。不少人对它的印象停留在“老古董”“被内存计算淘汰了”但只要你还在碰离线批处理、还在给上层 SQL 引擎做底层调优、还要处理几十 TB 的落地日志MapReduce 那套“切片、排序、归并、落盘”的路子就躲不开。它不只是一个框架更是一种把大问题拆成可并行小问题的思考方式。这篇东西我想把 MapReduce 从里到外拆一遍它在解决什么问题几个阶段各自干了什么shuffle 为什么是性能命门怎么写一个真能跑起来的作业以及我踩过的那些坑。适合刚入门、能看懂 Java 但没自己写过作业的同学也适合已经会用、但调优时心里没底的老哥。原理讲透代码给全参数给到具体数量级。1. 先把问题摆出来为什么单机一定扛不住1.1 一个 20TB 日志的尴尬现场假设你手上有一批按天滚动的服务端日志一天 20TB你要统计每个接口路径的调用次数、平均耗时、错误码分布。用最朴素的写法读文件、逐行解析、往一个哈希表里累加最后输出。单机跑瓶颈会依次撞上三堵墙。第一堵是磁盘 IO。机械盘顺序读算 150MB/s20TB 光读完就要 20×1024×1024/150 ≈ 14 万秒接近 39 小时这还是理想顺序读。第二堵是内存。去重后的接口路径轻松上千万条每条 key 加计数器按 200 字节估就是 2GB 常驻加上 JVM 对象头开销和 GC 压力很容易在几个小时后 OOM。第三堵最要命任何一步出错都得从头再来中间没有可恢复的检查点跑 30 小时崩在第 29 小时等于白干。分布式要解决的就是这三件事把 IO 压力摊到多台机器、把内存压力切成多份、把失败成本降到只重算一小块。听起来简单难的是怎么切、切完之后怎么合、合的过程中怎么保证不重不漏。1.2 分而治之不是口号是三次实实在在的切分很多人说 MapReduce 是分治思想这话没错但太虚。落到工程上它其实做了三次不同性质的切分这三次切分的对象和目的完全不同。第一次切分针对输入数据切出来的单位叫 split。一个 512MB 的文件按 128MB 切成 4 片每片交给一个 map 任务。这一步只记录“从第几个字节读到第几个字节”不复制数据所以切分开销几乎为零。切片数量直接决定 map 任务的并行度是并行度的源头。第二次切分针对中间结果的键空间切出来的单位叫分区partition。map 输出的是海量键值对按 key 的哈希值把它们分成 N 份每份最终交给一个 reduce。这一步决定了负载是否均衡是数据倾斜的根源所在。第三次切分针对单个分区内部的数据切出来的单位叫归并段。reduce 拉取自己那份数据后要先在内存里排序内存不够就溢写到磁盘形成一个个有序小文件最后做多路归并。这一步决定了单点内存压力和磁盘压力。这三次切分对应三类典型故障切片太少跑不满集群、分区不均导致长尾、归并段太多把磁盘打满。调优的时候先判断卡在哪一层比盲调参数有效得多。1.3 为什么核心链路坚持落盘而不是全程驻留内存这是 MapReduce 最被诟病的设计也是它最稳的地方。map 输出不直接推给 reduce而是先写本地磁盘reduce 拿到数据先落盘再归并。整个过程有大量磁盘往返速度上确实吃亏。但换个角度看内存驻留方案一旦某个节点挂掉丢失的是整条血缘链上的中间结果重算代价可能是整个上游而落盘方案里挂掉只影响这个节点上正在跑的那几个任务已经完成的 map 输出还在磁盘上只需重跑失败的那一小块。在一个由几千台普通机器组成、磁盘和网卡每天都在随机出问题的集群里这个取舍非常现实。我的经验是数据量在百 GB 级、迭代轮次多、对延迟敏感的活用内存计算确实快但数据量上到几十 TB、跑一次要几小时、中间还可能因为节点故障中断的场景落盘带来的可恢复性比那点速度值钱。理解这个取舍你就理解了 MapReduce 为什么长这样。2. 五个阶段拆开看数据在集群里到底走了哪几步2.1 输入格式与切片并行度从这里开始作业启动时客户端会先问输入格式要切片清单。默认实现按块大小切同时做一个小优化如果一片的剩余部分不超过块大小的 1.1 倍就整片给前一个 split避免出现极小的碎片片。这个 1.1 倍的阈值很多人不知道但它会实实在在影响你的任务数——100 个 130MB 的文件如果不做这个合并会切出 100 个 128MB 加 100 个 2MB 的小片多出 100 个几乎没有数据量的任务光调度开销就够呛。切片清单生成后每个 split 会被分配到一个任务上。分配的优先级是数据本地性最好分到数据所在节点同机其次同机架最后才是跨机架。跨机架传输要走汇聚层交换机带宽通常是机内的十分之一所以调度器会尽量等本地资源这个等待有超时超时后仍然会跨节点起任务不然小集群会饿死。注意切片是按字节切的不是按行切的。一个 split 的边界很可能落在一行中间读取时框架会跳过第一行的残片、并把最后一行读完这部分逻辑由 LineRecordReader 处理。你自己实现输入格式时如果忘了处理边界就会出现丢行或者重复行而且极难复现。2.2 Map 阶段能做和不该做的事map 函数的职责单一拿到一行输入吐出零到多个键值对。它的输入是(偏移量, 一行文本)输出是(任意可序列化的 key, 任意可序列化的 value)。有几件事我建议在 map 里做数据清洗和脏行过滤一条格式错乱的行直接 return比让它流到下游触发异常划算得多字段裁剪只保留后续要用的列中间结果能小一半轻量的维度关联把小的维度表通过分布式缓存加载到内存在 map 端就完成 join能省掉整个 shuffle。有几件事千万别在 map 里做初始化连接外部数据库或远程服务。map 任务可能成千上万每个任务建一次连接数据库连接池瞬间打满。正确做法是放到 setup 方法里并做懒加载或者更稳妥地放到每个 JVM 的静态块里因为一个 JVM 会复用跑多个任务。还有一件事是往 map 里塞超大对象JVM 堆是固定分配的塞爆了直接 OOM而 OOM 的堆栈信息往往指向别处排查起来很费劲。map 输出的键值对不会立刻落盘而是先写进一个环形内存缓冲区默认 100MB。当缓冲区用到 80% 时一个后台线程开始把数据排序后溢写到磁盘。这个比例调高能减少溢写次数但留给自己处理的空间就小了容易触发阻塞等待。2.3 Shuffle整个作业最贵的 70%如果只让我讲一个 MapReduce 的知识点我讲 shuffle。一个典型作业里shuffle 阶段消耗的时间经常占到总时长的 60% 到 70%剩下才是真正的计算。shuffle 分两半。map 端这一半缓冲区里的数据在溢写前会按 key 排序排序用的是分区号加 key 的复合比较器先按分区号排保证同一个分区的数据在文件里连续同分区内再按 key 排。这样 reduce 拉取时一个分区就是文件里一段连续的区间可以用一次范围读拿全。溢写文件如果有多个会在 map 结束前做一次多路归并合成一个带索引的大文件——索引记录的是每个分区在文件里的偏移量reduce 靠这个索引精确读取自己那份。reduce 端这一半每个 reduce 任务会起若干个拉取线程从所有已完成的 map 节点下载自己负责的那段数据。默认 5 个线程这个值对中小集群够用集群节点上千之后一定要往上调否则拉取会成为瓶颈你会在日志里看到大量“等待 map 输出”的提示。拉下来的数据先放内存放不下就溢写最后统一做归并排序把相同 key 聚在一起喂给 reduce 函数。这里有个容易被忽略的细节reduce 必须等所有 map 都完成后才能进入最终归并但拉取可以提前开始。只要某个 map 完成了reduce 就会去拉它的输出这叫 shuffle 的流水化。所以你会看到一个现象map 完成 95% 的时候 reduce 已经拉到 90% 的数据了最后卡在那 5% 的慢 map 上这段时间叫“长尾”。推测执行就是为这个设计的。2.4 Reduce 与输出归并之后才是业务逻辑reduce 函数接收的是(key, 该 key 的所有 value 的迭代器)。注意是迭代器不是集合框架为了省内存通常不会把这个 key 的所有 value 一次性加载到内存而是边归并边喂。这意味着两件事第一你不能对 values 做两次遍历第二次会拿到空第二你不能假设 values 的顺序如果业务依赖顺序得自己在 value 里带上排序字段处理。输出格式负责把结果写到目标存储。默认是写到文件系统的文本文件每个 reduce 任务产生一个文件。写 20 个 reduce 就会产生 20 个part-r-00000到part-r-00019的文件这也是为什么小文件问题在 MapReduce 里特别突出——下游如果再读这些文件输入切片又会碎一次恶性循环。最后说下计数器。框架内置了一堆指标读取的记录数、写入的记录数、溢写次数、拉取的字节数、失败的拉取次数。这些数字看着不起眼排查问题时是黄金线索。溢写次数异常多说明缓冲区太小或者数据分布有问题失败拉取次数高说明网络或者节点有问题map 输入记录数和输出记录数差距巨大通常是过滤逻辑写错了。3. 自己动手写一个能跑的作业3.1 环境与依赖本地验证用单机伪分布式最省事装好 JDK 8 以上和 Maven引入三个依赖就够common、client、mapreduce-client-core版本保持和集群一致版本不一致是最常见的“本地能跑集群报错”原因。dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-common/artifactId version3.3.6/version /dependency dependency groupIdorg.apache.hadoop/groupId artifactIdhadoop-mapreduce-client-core/artifactId version3.3.6/version /dependency打包时注意把依赖的 scope 设成 provided因为集群上已经有这些 jar打进去会让作业包从几 MB 膨胀到上百 MB每次上传都慢还可能出现类冲突。用 Maven 的 shade 插件打一个瘦包只带业务类和第三方小依赖是最稳妥的做法。3.2 Mapper 与 Reducer 的骨架拿一个按接口路径统计调用量的例子完整代码长这样public class ApiCallCounter { public static class TokenizerMapper extends MapperLongWritable, Text, Text, IntWritable { private static final IntWritable ONE new IntWritable(1); private final Text outKey new Text(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString(); String[] parts line.split(\t); // 字段不够直接丢弃别让一条脏数据拖垮整个 split if (parts.length 3) { context.getCounter(custom, bad_line).increment(1); return; } String path parts[1]; if (path.isEmpty()) { context.getCounter(custom, empty_path).increment(1); return; } outKey.set(path); context.write(outKey, ONE); } } public static class IntSumReducer extends ReducerText, IntWritable, Text, IntWritable { private final IntWritable result new IntWritable(); Override protected void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException { int sum 0; for (IntWritable v : values) { sum v.get(); } result.set(sum); context.write(key, result); } } }这段代码有两个细节值得说。一是复用对象outKey和ONE定义成复用实例而不是每次 new因为 map 可能被调用上亿次少创建上亿个短命对象GC 压力能明显降低。这是 MapReduce 编程里最值钱的一个小习惯。二是自定义计数器脏数据不是简单丢掉就完事要能看见丢了多少不然数据对不上时你会怀疑人生。3.3 Combiner、Partitioner 和驱动配置Combiner 是 map 端的预聚合本质上就是在本地的 map 输出上先跑一遍 reduce 逻辑。上面这个求和场景一个 map 输出 100 万条(path, 1)加了 combiner 之后可能只剩 5 万条(path, 累计值)网络传输量直接降一个数量级。job.setCombinerClass(IntSumReducer.class);但 combiner 不是随便加的。它要求操作满足结合律和交换律求和、求最大值、求并集可以求平均值不行——你会得到各分片平均值的平均而不是全局平均。这个坑我见过不止一个团队踩。真要算平均值就把 sum 和 count 一起输出最后在 reduce 里除。分区器决定 key 去哪个 reduce。默认用 key 的哈希值取模对绝大多数场景够用。但当你要保证同一类 key 落到同一个 reduce、或者要做二次排序时就得自己写public class PathPrefixPartitioner extends PartitionerText, IntWritable { Override public int getPartition(Text key, IntWritable value, int numPartitions) { String path key.toString(); // 按一级路径聚合保证 /api/user/* 全落到同一个 reduce int idx path.indexOf(/, 1); String prefix idx 0 ? path.substring(0, idx) : path; return (prefix.hashCode() Integer.MAX_VALUE) % numPartitions; } }注意 Integer.MAX_VALUE这一步不能省。哈希值可能是负数负数取模结果还是负数会直接抛数组越界而且只在特定 key 上偶发测试时未必能发现。驱动部分把参数串起来Configuration conf new Configuration(); conf.set(mapreduce.map.memory.mb, 3072); conf.set(mapreduce.map.java.opts, -Xmx2457m); conf.set(mapreduce.reduce.memory.mb, 6144); conf.set(mapreduce.reduce.java.opts, -Xmx4915m); conf.set(mapreduce.task.io.sort.mb, 300); Job job Job.getInstance(conf, api-call-counter); job.setJarByClass(ApiCallCounter.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setPartitionerClass(PathPrefixPartitioner.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); job.setNumReduceTasks(30); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 输出目录必须不存在否则直接抛异常这是新手最常见的第一个报错 System.exit(job.waitForCompletion(true) ? 0 : 1);内存参数有个固定的换算关系JVM 堆大小设为容器内存的 0.75 到 0.8。上面 3072MB 配 2457m 就是这个比例。堆设得和容器一样大进程会因为堆外内存被系统杀掉日志里通常只有一个含糊的“被信号杀死”查半天。3.4 reduce 数量怎么估参数怎么配reduce 数量是调优里最常被问的问题。我的估算方法分三步。第一步算单 reduce 处理的数据量。中间结果总量除以 reduce 数控制在 1GB 到 2GB 之间比较舒服。太少则 reduce 任务多、调度开销大、输出小文件多太多则单个 reduce 跑得久一旦失败重算代价高。第二步算中间结果总量。原始数据 10TBmap 端做了字段裁剪和过滤后可能只剩 3TB加了 combiner 之后可能压到 300GB。按 1.5GB 一个 reduce 算就是 200 个。第三步看集群资源上限。每个 reduce 占 6GB 内存集群总共可用的 reduce 槽位是 500 个那 200 个完全跑得开可以再调大到 300 缩短单任务时长。如果算出来 2000 个槽位只有 300就得砍到 300 左右多跑几轮。常见参数的参考值我整理成一张表都是实测比较稳的量级参数默认值建议值调整逻辑mapreduce.task.io.sort.mb100200~400缓冲区越大溢写越少但要给堆留余量mapreduce.map.sort.spill.percent0.800.80调到 0.9 以上容易阻塞写线程mapreduce.reduce.shuffle.parallelcopies520~50节点数上千必须调否则拉取成瓶颈mapreduce.reduce.shuffle.input.buffer.percent0.700.6~0.7太高会挤压 reduce 函数本身的堆mapreduce.job.reduces1按上式估算别用默认值1 个 reduce 是性能杀手mapreduce.map.speculativetrue小集群开大集群酌情长尾严重时开资源紧张时关mapmapreduce.reduce.memory.mb10244096~8192跟着中间结果大小走举个具体的溢写次数估算某个 map 输出 2GB 数据缓冲区 300MB溢写阈值 0.8那么每次溢写 240MB需要溢写 2×1024/240 ≈ 9 次最后再归并这 9 个文件。如果把缓冲区调到 100MB默认就是 25 次溢写加 25 路归并磁盘压力和耗时差出一截。这就是为什么我通常建议把io.sort.mb直接提到 300 左右。4. 排查实录那些让作业卡住的真实原因4.1 数据倾斜reduce 卡在 99% 的经典场景症状很典型30 个 reduce29 个在半小时内跑完剩下一个卡在 99% 加了两个小时还没动。打开这个 reduce 的日志会发现它在反复溢写说明单个 key 的数据量远超预期。根因通常是某个维度的值极度集中。比如按用户 ID 聚合有几个爬虫账号贡献了 40% 的记录或者 join 时某个默认值空字符串、-1、未知占了绝大多数它们全都哈希到同一个分区。我的处理顺序是这样的。先看能不能在 map 端消掉如果是大小表 join把小的那张表通过分布式缓存加载进内存map 端直接完成关联整个 shuffle 都省了这是收益最高的一招。如果两边都大就看能不能加 combiner 预聚合把重复 key 先在本地合掉。前两招都不行就上打散。给 key 加随机后缀比如key_0到key_9把原来一个 reduce 的活分给十个reduce 端拿到带后缀的 key 先做一次局部聚合再加一个只做汇总的第二个作业。两阶段聚合听起来麻烦但处理严重倾斜时是唯一稳的办法。还有一种容易被忽略的倾斜来源分区器写错了。我见过一个自定义分区器直接用key.hashCode() % numPartitions忘了处理负数结果一部分数据全跑到负索引上异常被吞掉后又重试表现就是某个 reduce 迟迟不结束。这类问题看计数器里的失败任务数能发现端倪。4.2 内存问题OOM 和 GC 是两条不同的线索OOM 分两种处理方式完全不同。一种是堆内存溢出日志里明确写着java.lang.OutOfMemoryError: Java heap space堆栈指向你的业务代码。这多半是在 map 或 reduce 里缓存了不该缓存的东西比如把 values 转成 ArrayList 再遍历、或者初始化了一个巨大的本地 Map。解决办法是改逻辑实在需要就调大堆但调大之前先想清楚数据量会不会继续涨。另一种是被系统杀掉日志里没有 Java 异常只有一个容器退出码 137或者一句“Container killed by the system”。这通常是物理内存超了堆外内存直接缓冲区、JVM 元空间、线程栈没算进去。这时候要做的是压低堆大小而不是调大很多人反着调越调越容易被杀。还有一种是 GC 时间过长被杀退出信息里会带“超出 GC 时间上限”。这一般是堆设得偏大但对象存活率高或者堆偏小导致频繁 Full GC。判断方法是看 GC 日志里 Full GC 后的老年代占用如果回收后仍然接近满就是真的放不下得改逻辑如果回收后降下来很多就是堆太小。4.3 小文件并行度不是越大越好新手的直觉是任务越多越快实际不是。假设输入是 10 万个 10MB 的小文件默认切片策略会切出 10 万个 map 任务。每个任务的启动开销JVM 启动、配置加载、连接建立大概 1 到 3 秒10 万个就是 30 到 80 小时纯启动开销还没算上调度队列的排队和 namenode 的元数据压力。解决办法有两层。第一层是入口合并用合并输入格式把多个小文件塞进一个切片按 128MB 的目标大小攒够再切任务数能直接降到千级。第二层是出口控制reduce 数量别拍脑袋定大如果最终结果只有 500MB用 100 个 reduce 就是 100 个小文件下游又得处理一遍用 5 到 10 个更合适。提示判断是否存在小文件问题看两个计数器——启动的 map 任务数和平均每个任务的输入字节数。后者如果远小于块大小比如只有几 MB基本可以确认是切片过碎。4.4 常见问题速查表现象大概率原因排查动作处理办法卡在 map 0% 很久资源排队或输入路径不存在看调度队列和路径配置等资源或修正路径map 长时间停在 99%单个 split 过大或数据本地性差看该任务的输入字节数调整切片大小拆分大文件reduce 卡在 33%/67%拉取阶段慢网络或 map 未完看拉取失败次数和剩余 map 数提高拉取并发处理慢 mapreduce 卡在 99%数据倾斜看各 reduce 的输入记录数加盐打散或改分区器容器被信号杀死物理内存超限对比堆设置和容器内存堆降到容器内存的 0.75输出目录已存在报错上次作业残留检查输出路径换路径或先删除结果条数比预期少脏数据被静默丢弃看自定义计数器补日志修清洗逻辑作业反复重试同一任务节点硬件问题或代码有非确定性 bug看失败任务的节点分布拉黑节点或修代码最后一行那个坑值得单独说一句。如果同一个任务在多个不同节点上反复失败那就不是硬件问题是代码里存在非确定性——比如依赖了系统时间、随机数、或者 HashMap 的遍历顺序。这类 bug 在单机测试时完全看不出来分布式一跑就现原形而且日志里往往只有一句“任务失败次数超过上限”非常有迷惑性。5. 这套模型今天还能给你什么5.1 它在现代数据栈里的真实位置直接写 MapReduce 作业的场景确实少了但它的模型没有消失而是下沉成了基础设施。上层那些用 SQL 描述的计算最终编译执行时聚合和 join 的物理算子仍然是“先局部聚合再全局归并”这一套分布式文件系统里的数据布局、分区表的目录结构、甚至列式存储里的行组划分本质都是切片的延伸。我个人的判断是不需要再手写大量 MapReduce 作业但必须懂它的执行模型。因为当你调优一个上层查询时你调的每一个参数背后对应的都是这套模型里的某个环节——并行度对应切片倾斜对应分区内存溢写对应缓冲区。不懂模型调参就是碰运气。5.2 三个能直接迁移到别处的设计习惯第一个习惯是先想清楚“怎么切”再想“怎么算”。不管是处理日志、批量生成报表、还是做图片批处理先确定切分维度和每片的处理成本比直接写循环要重要得多。切分维度选错了后面再怎么优化都是在错误的方向上努力。第二个习惯是把“局部预聚合”当成默认动作。能在数据产生的地方先合一次就别让它原样流到下一层。这个思路在缓存设计、接口聚合、日志上报里全都适用本质就是 combiner 的思想减少传输量永远比提升传输速度划算。第三个习惯是让每个环节都可重入。MapReduce 的任务失败能重跑是因为每个任务不依赖外部状态重跑结果一致。写批处理脚本、写数据同步逻辑时尽量保证“跑一次”和“跑三次”结果相同这个约束会逼着你把状态管理做干净长期看省下的排查时间远超当初多花的设计时间。真要说这套东西给我留下的最深印象是它教会我用“失败是常态”的视角做设计。单机思维里失败是意外分布式思维里失败是日常节点会挂、磁盘会坏、网络会抖把这些当成前提而不是异常设计出来的东西才经得起跑。调优这事儿没有银弹。我现在的习惯是每次作业跑完把计数器拉出来看一眼map 输入输出比、溢写次数、拉取字节数、失败任务数这四个数字基本能告诉你这次跑得健不健康以及下次该往哪个方向调。数据量涨十倍的时候这套观察方法比背参数表有用得多。
