如果你接手过那种对输出结果有强制拆分需求的离线任务一定对MapReduce的分区机制不陌生。前阵子我处理一个日志清洗任务数据量每天几亿条要按小时输出成24个目录给下游小时级报表使用。一开始图省事所有数据都丢给一个Reduce处理结果任务跑到天亮都没跑完后来把ReduceTask数量调大文件又乱成一锅粥根本没法按小时归类。直到认真跑了一遍Hadoop MapReduce分区案例才彻底明白问题出在Partitioner上。这篇文章就是我基于这个案例整理出的实践笔记。核心内容包括默认分区器有哪些局限性、自定义Partitioner怎么实现、分区编号与ReduceTask数量之间的关系以及数据倾斜的排查和缓解思路。如果你是刚接触MapReduce、或者已经写了几个WordCount但还没深入分区逻辑的Java开发者这篇文章应该能帮你省掉不少试错时间。1. 从一个实际需求说起为什么必须自己写分区逻辑1.1 默认分区器做了什么什么时候不够用MapReduce默认的分区器是HashPartitioner它的工作逻辑非常简单对Map输出key的hashCode取绝对值然后对ReduceTask数量求余得到这条记录该进入哪个分区。public class HashPartitionerK, V extends PartitionerK, V { public int getPartition(K key, V value, int numReduceTasks) { return (key.hashCode() Integer.MAX_VALUE) % numReduceTasks; } }这种设计的目标是让数据尽量均匀散开在大多数场景下没有问题。但一旦你有了明确的业务诉求比如“同一个小时间的日志必须进同一个文件”“同一月份订单必须落在同一个Reduce里统计”默认分区器就帮不上忙了。原因很简单默认分区器只关心key的哈希值不关心key的语义内容。同样是时间字段哈希值可能是乱序的同一个月的数据会被切得七零八落。我最初就是没有意识到这一点才导致改了ReduceTask数量后输出文件还是无法直接按小时使用。1.2 分区、分组和排序不是一回事很多人会把“分区”和“分组”搞混包括我自己早期也踩过这个坑。在MapReduce里这三个概念发生在不同阶段分区Partition决定Map输出的每一条记录被哪个ReduceTask处理发生在Map阶段的Shuffle过程中。排序Sort在分区完成后每个分区内部会按key做一次排序。分组Grouping在Reduce端key相同的记录会被合并到同一个Iterablevalue集合里这是reduce()方法能按key聚合的基础。所以分区解决的是“数据去哪个Reducer”分组解决的是“在同一个Reducer里数据怎么合并”。默认情况下它们是相互独立又有关联的只有先分区才能在一个分区内排序和分组。如果自定义了分区器你只是改了“去哪”的规则排序和分组规则并不受影响。这点理清楚之后自定义分区器的实现思路就豁然开朗了你需要做的事情只是重写getPartition()方法根据业务key计算出一个合法的分区编号。2. 一个可复现的月份分区案例从Mapper到Driver2.1 搞懂getPartition返回值的区间自定义分区器需要继承org.apache.hadoop.mapreduce.PartitionerK,V然后重写方法public abstract int getPartition(K key, V value, int numPartitions);注意三个参数key和value是Map阶段输出的key和value不是输入源文件的类型。numPartitions是当前作业设置的ReduceTask数量也就是执行时的分区总数。返回值代表“这条记录应该去第几个分区”编号从0开始最大值不能超过numPartitions - 1。一旦返回负数或者大于等于numPartitionsHadoop会直接抛出IllegalArgumentException因为下游没办法找到对应的ReduceTask来处理。我来拿一个实际案例讲解。2.2 一个按月份拆分的完整案例假设我们有某个电商平台一年内的订单明细数据每行格式如下2020-01-13,SKU1001,299.00 2020-02-03,SKU1002,59.90现在需求是按月份把订单数据拆成12个目录每个月一个文件并在Reducer里统计当月订单总金额。我们需要在Mapper里把日期截取成年月作为新的key输出然后自定义一个MonthPartitioner根据月份返回0到11之间的编号。Mapper实现import java.io.IOException; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; public class OrderMapper extends MapperLongWritable, Text, Text, Text { private Text outKey new Text(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString(); String[] fields line.split(,); if (fields.length 3) { return; } // 日期格式yyyy-MM-dd截取前7位得到yyyy-MM String month fields[0].substring(0, 7); outKey.set(month); context.write(outKey, value); } }自定义分区器import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Partitioner; public class MonthPartitioner extends PartitionerText, Text { Override public int getPartition(Text key, Text value, int numPartitions) { // key的格式是 yyyy-MM比如 2020-01 String monthStr key.toString().substring(5); int month Integer.parseInt(monthStr); // 返回 0 ~ 11 return month - 1; } }这里的key.toString().substring(5)取的是“01”到“12”这段字符串直接转成int并减一得到0到11的合法分区编号。Reducer实现import java.io.IOException; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class OrderReducer extends ReducerText, Text, Text, Text { Override protected void reduce(Text key, IterableText values, Context context) throws IOException, InterruptedException { double total 0.0; for (Text value : values) { String[] fields value.toString().split(,); if (fields.length 3) { total Double.parseDouble(fields[2]); } } context.write(key, new Text(String.valueOf(total))); } }Driver主类import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; 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 OrderMonthDriver { public static void main(String[] args) throws Exception { if (args.length ! 2) { System.err.println(Usage: OrderMonthDriver input output); System.exit(-1); } Configuration conf new Configuration(); Job job Job.getInstance(conf, order month partition); job.setJarByClass(OrderMonthDriver.class); job.setMapperClass(OrderMapper.class); job.setReducerClass(OrderReducer.class); // 关键1指定自定义分区器 job.setPartitionerClass(MonthPartitioner.class); // 关键2设置ReduceTask数量为12对应12个月 job.setNumReduceTasks(12); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }2.3 提交作业时的两个关键配置第一个是job.setPartitionerClass(MonthPartitioner.class)这是让Hadoop使用自定义分区器的入口。第二个是job.setNumReduceTasks(12)它决定了分区的总数量。这两个配置必须搭配好。因为getPartition()方法里的numPartitions参数就是setNumReduceTasks()里设置的值。如果你设置12个Reduce那么分区编号只允许在0到11之间选择如果你设置10个Reduce而业务仍然坚持12个分区month - 1在最后两个月就会返回11和10不对month-1对于month11返回10对于month12返回11而numPartitions10时允许的最大编号是9因此第11月和第12月对应编号10和11都会超出合法范围。我把这个坑放在下一节专门讲因为这是最容易忽视、报错又很隐晦的地方。3. 分区编号、ReduceTask数量和输出文件的三角关系3.1 分区编号的唯一合法区间是[0, numPartitions-1]这是整个分区机制里最基本的约束。numPartitions由numReduceTasks决定getPartition()的返回值必须落在[0, numPartitions-1]区间内。为什么必须是这个区间因为Hadoop在Shuffle阶段需要把每个Map输出按分区编号写到对应的临时文件并在ReduceTask启动后根据编号从各个Map拉取属于自己那份数据。如果编号超过范围Hadoop根本不知道这个分区对应哪个ReduceTask只能抛异常终止作业。我在初次实现分区时为了省事直接返回了月份原值month也就是1到12。当时setNumReduceTasks(12)结果1月数据进入分区1没问题但12月数据进入分区12而分区编号只有0到11于是作业运行到一半就报错。这类错误在日志里通常表现为java.lang.IllegalArgumentException: Partition out of range: 12排查起来并不难难的是很多人根本没意识到这个返回值范围限制。3.2 ReduceTask数量小于业务分区数会报错而不是自动合并有一种常见误解是如果我只设置10个ReduceTask但分区逻辑返回0到11是不是后两个月的数据会自动合并到第10个Reduce里不会。MapReduce的分区是精确匹配不是区间折叠。Partitioner没有做任何“如果编号超出范围就取模”的兜底逻辑。你返回0到11但只开10个ReduceTask编号10和11就是非法分区作业直接失败。所以在设计时你要么让getPartition()的返回值永远小于numPartitions要么确保numPartitions足够覆盖业务分区数量。一个稳妥做法是在getPartition()里对返回值做一个保护性约束比如int result month - 1; return result % numPartitions;这样即使numPartitions设置小了也不会直接报错而是取模合并到已有分区里。但要注意这么做可能把原本希望分开的数据又合并了所以保护性取模只能作为兜底不能真正替代“合理设置ReduceTask数量”。3.3 ReduceTask数量大于业务分区数会多出空文件反过来如果业务只需要12个月但你把setNumReduceTasks(24)会发生什么呢作业不会报错但会有12个分区编号0到11有数据另外12个分区编号12到23没有任何记录。最终输出目录里会出现24个结果文件其中大约一半是空的。空文件在HDFS上是真实存在的文件大小可能为0也可能只有文件头信息。这个场景常见于“想用更多Reduce来加速”的场景。如果业务确实只需要12个月多出来的Reduce不仅浪费计算资源还会让下游处理多出12个空路径导致文件解析逻辑报错或者白白多跑很多空任务。我当时就在这个细节上栽过跟头。所以我的经验是先明确业务需要的分区粒度再决定setNumReduceTasks的大小。如果只是为了并行加速不要和业务分区数量脱节最好把numReduceTasks设置成业务分区数的整数倍并在getPartition()里对编号取模让每个业务分区对应多个子分区。4. 分区不均匀导致的数据倾斜识别与缓解4.1 数据倾斜的现场长什么样结构上分区逻辑没毛病不等于性能没问题。我第一次把订单按月份分区跑全量数据时发现前11个月的任务很快跑完但12月那个ReduceTask卡了快两小时。日志里其他Reduce任务早就进入100%只有它还在慢慢吞吞地写入。这就是典型的数据倾斜数据在分区之间分布不均匀导致某个Reduce处理的数据量远大于其他Reduce。原因不外乎几种分区键本身分布就不均衡比如某个月是大促月份订单量是平时的几倍。分区粒度太粗一个月就一个分区无法拆分热点月份。分区逻辑过于简单只按业务标记分没有考虑底层数据规模。对于离线跑批来说数据倾斜不一定会让作业失败但会显著拉长整体运行时间还可能在Reducer端触发内存溢出。4.2 先量化分区键分布再写分区逻辑很多人在写分区器时凭直觉定义业务分区键却忽略了对数据分布的先期探查。我后来养成了一个习惯在写自定义Partitioner前先对输入数据做一次采样统计看看每个分区键大概有多少条记录。在Hadoop环境里最简单的做法是跑一个轻量级的MapReduce任务输出分区键以及它们的计数。比如针对订单数据可以先统计每个月出现的次数public class MonthCountMapper extends MapperLongWritable, Text, Text, LongWritable { // 从日期字段截取月份输出 (月份, 1) }拿到这个分布后你就能直观看到哪个月的数据量特别大。如果某个月明显偏多那就得重新设计分区策略而不是直接把几个月合并到一个分区里。4.3 复合分区键和二次分区思路想要缓解倾斜最笨但有效的办法是“细分”分区键。比如原本按月份分区如果发现12月数据量是其他月份的5倍可以把12月按天或者按商品品类继续细分让多个Reduce共同承担这个热点月份的压力。这时你可以在Mapper里把输出key改成组合键比如“月份-品类”或者“月份-日期”在Partitioner里根据组合键的某一部分计算分区编号再在Reducer里针对这个细分粒度进行聚合。public class CompositePartitioner extends PartitionerText, Text { Override public int getPartition(Text key, Text value, int numPartitions) { // key格式: 2020-12-GROUP_A String[] part key.toString().split(-); // 用月份哈希和品类哈希混合计算 int monthHash part[0].concat(part[1]).hashCode() Integer.MAX_VALUE; return monthHash % numPartitions; } }这种做法的代价是你可能增加了ReduceTask数量最终输出文件数量也变多需要下游做一层合并。但相比让某个Reduce单点跑爆这个代价是值得的。还有一种思路是“二次分区”先用默认分区器把数据散开然后在Reducer内部做二次归类。但这需要你在Reduce阶段手动维护多个输出流代码会复杂不少适合输出文件数量可控的场景。5. 跑批过程中的高频坑和验证方法5.1 自定义分区类没被打进提交的jar包很多人遇到过这种情况本地测试好好的丢到集群上就报ClassNotFoundException: xxx.MonthPartitioner。原因通常是提交作业时自定义分区器的类没有被打包进最终提交的jar。如果你用Maven打包最好确认maven-jar-plugin包含了所有依赖类和自定义类。如果你是在一个工具类比较多的工程里开发更要注意很多打包插件默认只打主模块的类分包里的类会被遗漏。我自己的检查方式很简单提交前先解压jar包看看里面是否存在MonthPartitioner.class。没有的话直接重新打包。5.2 部分输出文件为空别急着改代码刚跑通分区案例时看到输出目录里一堆空文件第一反应是代码写错了。但其实这在“ReduceTask数量大于业务分区数”时是正常现象。要判断是不是正常可以在Driver里打印一份统计数据或者在任务运行完成后用hdfs dfs -ls查看每个输出文件的大小。如果空文件数量恰好等于numReduceTasks - 业务分区数那基本可以确定是ReduceTask设置多了。如果空文件数量对不上再检查Partitioner返回值的区间是不是有黑洞。5.3 用Counter快速统计每个分区记录数还有一种更优雅的验证方式利用Hadoop的Counter机制在Map或Reduce阶段给每个分区编号打点。比如在自定义Partitioner里无法直接访问Counter但可以在Reducer里通过context.getCounter(partition, String.valueOf(分区编号))来计数。具体做法是在Reducer的setup()里拿到当前Task的上下文再根据context.getTaskAttemptID()解析出当前Reduce编号在reduce()方法里递增对应的Counter。任务跑完后从控制台日志或者Web UI的Counters列表里就能看到每个分区实际处理了多少条记录。如果发现某个分区Counter远高于其他分区说明分区分布不均需要回头检查分区键设计。这个方法比看输出文件大小更直接尤其在数据压缩或文件头信息影响文件大小判断时。最后再分享一个我自己常用的调试小技巧写自定义Partitioner时别一上来就处理十几亿条生产数据先拿一个小文件在本地模式跑通通过输出文件和日志验证分区逻辑无误后再扔到集群上跑全量。这个习惯帮我省下的排查时间远超那些本地初始化的几分钟——毕竟跑一个失败的全量任务通常要等十几分钟甚至更久。