我当年第一次把MapReduce作业从单机演示数据切到生产环境真实数据时最先跪的不是业务逻辑而是一个平时根本没人注意的组件——分区器Partitioner。日志倒是报错但排错排了很久才意识到所有数据都被发到了同一个Reduce任务另外几个Reduce干等整个作业慢得像一锅粥。后来我养成了一个习惯每看到一个MapReduce作业先问自己三个问题——数据按什么Key分区同一个业务维度的数据能不能落到同一个Reducer分区边界和排序顺序是不是一致这三个问题全部跟Partitioner有关。这篇内容不是那种“认识一下分区器有几种”的科普而是实打实地把分区器的源码、自定义方式、与排序分组的协作关系、以及运行期各种诡异故障一起讲透。不管你是正在刷MapReduce排序、自定义排序、分组排序实训题的同学还是想弄明白网约车数据清洗、招聘数据清洗这类综合项目里“为什么这样分区”的同行都应该能从中捞到点能直接用的东西。1. 分区器到底在解决什么问题1.1 数据流动的必经之路在MapReduce里Map任务和Reduce任务不是直接点对点通信的。Map端把海量(key, value)算出来之后下一步不是直接发给某个Reducer而是先要经过一个“打标签”的过程每条记录需要确定自己该去哪个Reduce任务。这个打标签的动作就是Partitioner的getPartition()方法。整个过程实际上是map()函数输出一条记录框架立刻调用分区器算出这条记录对应的分区号这个分区号会和(key, value)一起被写进环形缓冲区之后溢写、排序、合并等操作都以“分区”为基本单位展开。换句话说分区号是跟着记录一起走的“目的地地址”后续所有Shuffle过程都依赖这个标签。很多同学容易忽略一个关键点Map端在溢写时并不是单纯按Key排序而是先按分区号分组再在每个分区内按Key排序。这个顺序非常重要它决定了同一分区内的数据天然有序但不同分区之间没有大小关系。理解了这个顺序后面再看全排序、分组聚合一类问题会轻松很多。1.2 分区的本质提前给数据定好归属分区的本质是提前在Map端给每条记录决定“归属地”。Reduce任务是可以重启、可以推测执行的但一旦数据被发送到错误的Reducer后面再怎么重试都救不回来。可以打个比方分拣中心里每个快递包裹到了传送带分拣员只看地址也就是PartitionKey就决定把它扔进哪个格子这个格子对应哪个区域的分拣口包裹如果扔错格子就算后续运输环节再完美最终也到不了正确的人手里。MapReduce里也是这样Partitioner返回的整数就是“格子编号”范围是从0到numReduceTasks - 1。同一个分区编号的数据会全部进入同一个Reducer。而你真正关心的是业务逻辑同一台网约车订单能不能汇总到一起同一个城市的招聘数据能不能写进同一个目录同一批异常日志能不能单独归一个Reducer处理这些“能不能”全看分区器有没有把对应Key归到同一个分区。1.3 分区不只是负载均衡我见过不少文章把Partitioner简单描述成“让数据均匀分布到各Reducer”这其实是个误导。负载均衡只是分区的副产品而且很多时候根本做不到均衡。分区的核心目标是“数据逻辑归属正确”也就是把业务上需要一起处理的数据放到同一个Reduce里。举例来说假设你有四个Reducer想按城市统计招聘数据北京的数据必须全部进入同一个分区哪怕北京的数据量占全量的一半。如果强行均匀打散每个Reducer都拿不到完整的北京数据统计结果必然是碎的。所以说分区器关心的是“哪些数据该在一起”而不是“每份数据是否一样多”。数据倾斜问题不能只靠改分区器彻底解决它是另一个维度的工程问题这一点后面会展开讲。2. 默认分区器的源码级拆解2.1 HashPartitioner 的一行代码Hadoop默认的分区器是HashPartitioner它的核心代码就一行在org.apache.hadoop.mapreduce.lib.partition.HashPartitioner里public class HashPartitionerK, V extends PartitionerK, V { public int getPartition(K key, V value, int numReduceTasks) { return (key.hashCode() Integer.MAX_VALUE) % numReduceTasks; } }这一行代码里其实藏着三个值得留意的点。第一key.hashCode()调用的完全是Java的Object哈希逻辑对字符串、数字、自定义对象都用各自的hashCode()。这意味着如果自定义Key的hashCode()方法写得不好默认分区的均匀性就会很差。第二 Integer.MAX_VALUE的作用是把哈希值的符号位抹掉确保结果是非负整数。因为hashCode()可能返回负数负数取模结果也是负数而分区号不允许是负数。第三取模% numReduceTasks意味着即使两个Key的原始哈希值差别很大只要取模结果相同它们就会被分到同一个分区。所以默认分区只能保证“同一个Key一定到同一个Reducer”不能保证“不同的Key一定到不同的Reducer”。2.2 什么场景用默认分区就够了如果业务只需要“散列均匀”默认的HashPartitioner其实很省心。比如单纯做WordCount你统计每个单词的次数Reduce需要做的是按Key合并不关心某个单词和另一个单词是否在同一个Reducer只要相同的单词能聚到一起就行。这种情况下默认分区器配合合理的Reduce数量完全够用。再比如跑简单的ETL清洗清洗逻辑本身不依赖数据顺序也不要求某类数据一定放在某个Reducer那么默认分区器基本不会出问题。很多实训平台上的基础编程题其实用默认分区器就能过因为判题更关注的是Map和Reduce里的逻辑是否正确。但要注意默认分区器只管“同Key同Reducer”它不会帮你实现“同业务维度同Reducer”。比如你想把所有“北京”的数据汇总到一个Reduce但Key是完整的招聘信息字符串而不是“北京”这个城市名那么即使两条记录都来自北京它们的完整字符串hashCode也大概率不一样取模之后更可能被分到不同分区。这种需求默认分区器就不够了。2.3 哪些信号说明必须自定义了出现下面这些情况就需要考虑写自定义Partitioner了这些信号我基本都踩过需要把某个业务维度的数据聚到同一个Reducer但分区Key并不是当前Map输出Key本身而是Key里的某个字段。希望某些特殊数据比如脏数据、异常记录、超长文本单独进一个Reducer做专项处理。要求输出结果全局有序且Reduce任务数大于1这时候默认Hash分区会把Key打散全局有序根本无从谈起。数据倾斜严重默认哈希对业务型Key无能为力需要考虑按区间或采样边界分区。网上那些“自定义排序”“分组排序”实训题卡点很多时候其实不是排序Comparator而是Partitioner同一个业务组的数据如果被拆到不同Reduce后面怎么写Comparator都是白搭。3. 三种常见的自定义分区实战3.1 自定义Partitioner的骨架与三个关键点新API里自定义分区器只需要继承PartitionerK, V实现getPartition()方法即可。骨架代码大概是这样的public static class CityPartitioner extends PartitionerText, Text { Override public int getPartition(Text key, Text value, int numReduceTasks) { String city key.toString().split(\\|)[0]; if (beijing.equals(city)) { return 0; } else if (shanghai.equals(city)) { return 1; } else if (guangzhou.equals(city)) { return 2; } return 3; } }这里有三条经验每一条都是用报错换来的。第一返回值必须落在[0, numReduceTasks - 1]区间。如果返回了负数或者大于等于numReduceTasks的值作业会直接报Illegal partition之类的错误。尤其要注意如果你先写了固定返回3的分区器后来把Reduce数量改成了1分区器返回3就直接炸了。所以代码里一定要根据numPartitions做一次兜底。第二多个不同的Key落在同一个分区是完全正常的。分区的粒度是业务维度不是Key本身。比如多个城市可以通过路由逻辑映射到同一个分区完全没有问题。第三numPartitions参数就是运行时Reduce任务的数量。如果你调用了job.setNumReduceTasks(4)这个参数就是4。不要在Partitioner内部写死某个值来忽略这个参数否则Reduce数量一变代码就出问题。3.2 案例一招聘数据清洗按城市分区很多综合实训里都有“招聘数据清洗”这道题数据字段一般是这样城市、职位、公司、薪资下限、薪资上限、发布时间。需求是把不同城市的招聘数据分别清洗、去重最后每个城市单独输出一份文件方便后续按城市分析。如果直接用默认HashPartitioner城市字段只是整个Key的一小部分Map输出的完整Key可能是一整行数据不同城市的记录会被打散到各个Reduce。所以第一步就是自定义分区器按城市名来定位public static class RecruitPartitioner extends PartitionerText, Text { private static final MapString, Integer CITY_INDEX new HashMap(); static { CITY_INDEX.put(北京, 0); CITY_INDEX.put(上海, 1); CITY_INDEX.put(广州, 2); CITY_INDEX.put(深圳, 3); } Override public int getPartition(Text key, Text value, int numReduceTasks) { String[] fields key.toString().split(\t, -1); String city fields[0]; Integer idx CITY_INDEX.get(city); if (idx ! null) { return idx; } return numReduceTasks - 1; // 其他城市统一进最后一个分区 } }同时设置job.setNumReduceTasks(5)前四个分区放四个一线城市最后一个兜底放其他城市。这里有个细节如果实际城市数量超过分区数量那些城市就会挤到兜底分区里后续再细分就需要先跑一个统计任务拿到全部城市列表再动态构造分区映射。对于实训题来说固定映射通常够用。3.3 案例二网约车综合项目中的多维度分区网约车大数据项目场景更接近生产每天几千万条订单记录要按城市做数据清洗、统计时长、金额等指标。这时候需求往往不只是“按城市分区”还要“同城市内再按日期或时段组织”。但如果Map输出Key直接设计成城市_日期的组合字符串默认HashPartitioner会把同一个城市的不同日期拆到不同分区你没法在一个Reducer里得到整个城市的数据。更合理的做法是自定义一个组合Key对象并且巧妙利用“分区、排序、分组”三个环节的配合。思路是这样Map输出Key用CityDateKey包含城市和日期自定义Partitioner只根据城市字段决定分区自定义排序比较器首先按城市排序再按日期排序自定义分组比较器只根据城市分组。这样同一个城市的所有数据会进入同一个Reducer并且在Reducer内部这个城市的数据又会按日期排好序reduce()每次被调用时拿到的就是“一个城市、所有日期有序”的数据集。这个案例也是网上“自定义分组”题目的核心逻辑。很多人以为自定义分组只需要写一个GroupingComparator就够了实际上如果Partitioner不配合分组比较器根本发挥不出来。三者是联动的缺少任何一环都会出问题。3.4 案例三倒排索引和自定义排序任务中的分区策略“倒排序索引”也是实训题里高频出现的名字。它的典型需求是给定若干文档输出每个词条以及它出现过的文档编号列表并且文档编号有序。比如hadoop doc1, doc3, doc5 spark doc2, doc4如果词条数量很大需要多个Reducer分担那你需要对词条做分区。最简单的方式是默认HashPartitioner但由于倒排索引最终希望词条输出有顺序默认分区并不能保证“字典序靠前的词条一定落在编号靠前的输出文件里”。所以这道题的正确套路是要么直接把Reduce数设为1让一个Reducer处理所有词条内部用排序Comparator排字典序这样输出文件是全局有序的要么用采样方式确定词条边界做RangePartitioner这是进阶做法。如果你只需要完成实训题把Reduce数设为1是最省事的做法。但如果在生产环境单Reducer吞吐量不够就必须用RangePartitioner的思路这也引出了下一节要说的内容分区和排序的关系。4. 分区、排序、分组谁都别越权4.1 三者执行的先后顺序这是我在讲MapReduce时最想让大家记住的一张“流程表”环节职责本质Partitioner分区器决定每条记录进哪个Reducer数据归属Sort排序器决定Reducer内部按什么顺序处理数据顺序GroupingComparator决定调用几次reduce()、每次传入哪些Key数据分组边界在Map端流程是先分区、后分区内排序、再合并溢写文件。在Reduce端流程是拉取本分区数据、合并排序、按分组比较器把相同分组的Key合并成一组、调用一次reduce()。有一个很常见的理解误区认为排序发生在Reduce拉取数据之后。其实Map端溢写时就已经在做局部排序了Reduce端的排序主要是归并Map端的多个有序片段。但无论哪种排序都只是在“分区内部”进行的跨分区的数据之间没有任何大小关系。这个认知必须刻在脑子里。4.2 “全排序”问题的正确打开方式如果你要求所有Reduce输出合在一起是全局有序的那么分区边界必须和排序边界一致。默认HashPartitioner做不到因为两个Key的哈希顺序和它们的字典序没有任何关系。正确的做法是采样之后做RangePartitioner先对Key集合采样得到一批有序的“分割点”Partitioner根据Key落在哪个分割区间返回对应的分区号。比如采样得到的分割点是apple, banana, cherry那么字典序小于apple的去分区0apple到banana之间的去分区1以此类推。这样分区内再排序整个输出合起来就是全局有序的。Hadoop里提供了TotalOrderPartitioner和InputSampler可以帮我们自动完成这步。在旧版本里使用TotalOrderPartitioner.setPartitionFile(jobConf, partitionFile)配合InputSampler.writePartitionFile(job, sampleInput)完成在新API中也有对应工具类。不过很多人容易踩一个坑生产数据分布不均匀时单纯按采样边界分区仍然可能出现数据倾斜比如大量Key集中在某个区间。这就要根据实际分布做加权采样或者接受一定程度的倾斜再配合其他手段。4.3 排序分组实训题里为什么第一步是分区那些“第1关MapReduce排序—倒排序索引”“第1关自定义排序”之类的题目本质上都在考察“如果你连分区都没设计对后面的排序分组全都白搭”。很多同学看完题目就去写Comparator写完发现输出顺序不对或者同一组数据被拆得到处都是其实问题往往出在Partitioner。举一个最简单的例子需求是统计每个城市招聘岗位的平均薪资并且输出要按“城市平均薪资”组合排序。这时Map输出Key如果是城市那默认HashPartitioner其实够用但如果Map输出Key是城市|薪资这种组合字段而你的排序要按薪资排那么Partitioner如果再按完整的组合字段哈希就会把同一城市的数据分到不同分区结果每个Reducer拿到的都是碎片数据统计口径直接就错了。正确做法是把完整Key写成一个自定义WritableComparable同时让它的hashCode()只基于城市字段计算或者干脆在Partitioner里面解析出城市字段决定分区。这样同一城市一定同区排序比较器又可以在区内按完整Key排序需求才能同时满足。这个坑在真实项目中太常见了我建议所有做自定义Key的同学都回头检查一下自己的hashCode()到底写的是什么。4.4 分区键和排序键不一致的坑顺着上面那个例子继续说。当你使用组合Key时hashCode()、compareTo()、GroupingComparator三者分别服务于分区、排序、分组完全可以各不相同。但很多人会无意中把它们写成一致结果现象很诡异Point A代码看起来没问题跑出来分组却是乱的。有时候问题就出在这个细节自定义组合Key对象的hashCode()直接用了IDE自动生成的代码把城市、日期、薪资、ID全部混进哈希计算。于是两个只有薪资不同的记录城市完全相同哈希值却不相同默认HashPartitioner可能把它们分到不同分区。同一城市的记录被拆到不同分区后哪怕你的GroupingComparator只比较城市也没用因为分组比较器的前提是这些数据已经在同一个Reducer里了。解决方案是要么自定义Partitioner显式用城市字段分区要么让组合Key的hashCode()只依赖业务分区字段。我推荐前者因为代码意图更清晰也不会影响到其他可能复用到该Key的地方。5. 运行期那点事分区相关故障与排查5.1 任务失败Illegal partition最常见的错就是自定义Partitioner返回值越界。报错信息里通常会出现Illegal partition for之类的内容然后在日志里能看到预期的分区数范围。排查方法很直接先看numReduceTasks是几再去Partitioner里检查所有return路径的返回值范围。曾经有个同事写的分区器在if-else后面忘了加兜底某个不知名城市走进来之后返回了-1作业直接挂了。后来我要求团队里所有自定义Partitioner默认分支必须返回numPartitions - 1而且要确保numPartitions不是0。提示如果你在作业中途修改了Reduce数量比如从5改回1但分区器代码里还写死了返回4那么这个作业100%会失败。改Reduce数量时永远要同步检查Partitioner。5.2 输出文件数量对不上有同学问我明明设置了job.setNumReduceTasks(5)为什么输出目录里只有part-r-00000一个文件这个问题需要分层排查。第一层确认设置有没有真正生效。检查job.waitForCompletion(true)日志看Reduce任务数量是不是5。有时候配置被后续代码覆盖比如有人重新调用了job.setNumReduceTasks(1)或者读取了一个旧的配置文件。第二层确认最后一个Reduce是不是真的没收到数据。如果Partitioner实现里某些条件把所有数据都导向了分区0那么其他4个Reducer虽然启动了但没有任何输入自然也不会生成输出文件。第三层如果用了自定义分区的子类但忘记在Job里设置job.setPartitionerClass(...)框架会用默认HashPartitioner输出分布就和预期完全不同了。这类问题不一定是代码逻辑错往往是“设了但没设对地方”。5.3 数据倾斜从分区角度如何定位数据倾斜是另一个高发问题但很多人不知道要从分区工具去定位。当某些Reducer长时间运行、另一些Reducer几秒就结束时不要急着去调堆内存先看数据到底是怎么分布的。最直接的办法是在作业完成后查看Counter重点关注每个Reducer的输入记录数。命令行可以这样看mapred job -counter job_id org.apache.hadoop.mapreduce.TaskCounter REDUCE_INPUT_RECORDS如果某个Reduce的输入记录数明显比其他多一个数量级那说明分区策略把大量数据聚到了同一个分区。这时候要问自己这个分区对应的业务维度是否天然就大比如网约车项目里北京、上海的单量就是比三线城市大得多这是客观存在的倾斜。应对办法通常有两种一是把分区的粒度降下来比如从“按城市”改成“按城市时间段”让每个Reduce处理的数据量更均衡二是保留业务分区但把热点数据拆成多个子分区再使用MultipleOutputs输出到同一业务目录这样既保住了逻辑归属又缓解了计算倾斜。5.4 让Partitioner自己“说话”还有一个我觉得特别实用的排查技巧给自定义Partitioner加上调试输出或者Counter。虽然Partitioner是Map端执行的组件但它里面写的Counter会计入作业的总Counter可以用来确认每条记录的分区结果是否符合预期。比如在getPartition()里根据分区号做累加if (numPartitions 0 result numPartitions - 1) { // 这里可以增加一个自定义计数器统计兜底分区的记录数 }但要注意不要在生产环境频繁打印日志因为Partitioner在Map端会被并发调用日志量会非常大。合理的做法是抽样打印比如每执行到第1000次才打一条日志既能看到趋势又不至于淹没日志系统。我在实际工作中还有一个习惯任何涉及自定义分区的作业我第一次跑通后一定会故意把Reduce数量改成“分区数1”或者“分区数-1”看作业会不会报错。这个小动作能快速验证Partitioner是否真的具备对numPartitions的兼容性。很多同学写代码只针对一种配置一旦生产环境调整并行度就炸这种“配置变化测试”能提前筛掉一大批隐患。最后再分享一个观察网上大把MapReduce排序、分组、倒排索引的实训题大家卡住的位置惊人地一致——不是不会写Comparator而是没想过Partitioner才是那个支配全局的角色。如果你也遇到“排序结果不对”“分组结果拆散”“某个Reduce忙死其他Reduce闲死”这类问题不妨先翻一翻自己的分区器。十有八九一查一个准。
