1. 分区机制与设计思路
这次来聊一个老生常谈但很容易搞出玄学效果的话题：Hadoop MapReduce里的分区（Partition）机制。尤其是做Java开发的同学，写完Mapper和Reducer可能觉得万事大吉，结果跑出来的结果文件要么只有两个，要么一堆小文件，要么数据全堆到一个文件里，根本没法用。这些现象背后十有八九就是分区逻辑没吃透。
分区这个概念，放在MapReduce里其实非常直白：Map阶段处理完数据之后，会产出大量键值对，这些键值对怎么分配给不同的ReduceTask去处理，就是分区说了算。默认情况下，Hadoop用的是HashPartitioner，它做的事情是拿key的哈希值对ReduceTask数量取模，然后把记录分到对应的分区里去。这样做的目标很单纯——让每个Reducer拿到的数据尽量均匀，别让某个节点累死，其他节点闲死。
但问题来了，默认分区只看key的哈希，它根本不关心你的数据到底长什么样。假设你的key是手机号，你想让同一归属地的手机号写到同一个结果文件里，HashPartitioner做不到。又或者你的key本身是个分类标签，比如"北京""上海""广州"，你希望每个标签单独一个输出，HashPartitioner也只是碰巧能做到一部分，一旦标签数量变化，结果就乱套。
所以自定义分区的核心价值，就是用业务逻辑替代默认的哈希逻辑，让数据按照你的规则去分桶。这不只是输出文件的整理问题，它直接影响后续的负载均衡、全局排序、以及数据聚类的效率。很多初学者把它当成一个"锦上添花"的功能，其实它是MapReduce框架里少数几个能同时影响性能和结果形态的关键环节。
适合看这个案例的人，我大致分三类：一是刚学完MapReduce基础、想搞懂分区到底怎么生效的；二是工作中遇到"结果文件分布奇怪""ReduceTask设了没用"这类问题想排查的；三是准备面试、需要把分区机制讲清楚并动手实现过的人。这个案例我会从原理到代码一步步拆，保证你读完能直接照着写。
1.1 从默认分区说起
很多人写MapReduce的时候根本没主动设置过区，但程序照样跑。原因就是Hadoop给了你一个兜底的默认实现。在Job没有调用setPartitionerClass的时候，框架会使用org.apache.hadoop.mapreduce.lib.partition.HashPartitioner。
它的getPartition方法就一行逻辑：
return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;这里有个细节值得注意：它先用Integer.MAX_VALUE做了一次按位与。为什么要这么干？因为hashCode()可能返回负数，而取模的结果如果是负数，分区编号就会是负的，后续写数据的时候就会报错。所以先把你变成非负数，再取模。
这个默认逻辑的优点是简单、均匀，只要key的哈希分布够散，数据基本能平均分到各个Reducer。但缺点也很明显：它完全无视业务语义。你想让相同属性的一类数据进同一个Reducer，它做不到；你想控制每个Reducer处理的数据范围，它也做不到。更头疼的是，当你把ReduceTask数量调大以后，同一类数据可能被拆到多个Reducer里，后续做Global Sort或者按类聚合的时候就彻底乱了。
1.2 为什么要自定义分区
自定义分区的核心目标就一句话：把业务规则翻译成分桶规则。我给你举一个实际例子，也是这个案例里要做的场景。
假设你现在有一批手机用户的流量日志，字段包括手机号、上行流量、下行流量、总流量。你的需求是：按照手机号前三位来区分归属地，相同前三位的数据输出到同一个结果文件。这样后续做区域运营分析时，直接拿文件就能用，不需要再全量扫描重新归类。
默认分区做不到这一点，因为手机号整体作为key做哈希，产生的分布完全是随机的。即使你把ReduceTask数量设成和归属地数量一样，也无法保证某个手机号前三位固定落在某个分区里。
再比如，你的数据有冷热之分，想让高频key进同一个Reducer做合并，低频key进另一个Reducer，默认的分区策略也做不到。总之，一旦你开始关心Reducer输出的内容和结构，分区就需要自己说了算。
自定义分区的方法很简单：继承Partitioner抽象类，重写getPartition方法，然后在Driver里设置job.setPartitionerClass。真正有难度的不是代码写法，而是明白你的分区逻辑要跟ReduceTask数量配合好，否则分区写了也白写。
提示：分区的编号决定了ReduceTask的编号。分区编号从0开始，必须连续且小于ReduceTask数量。如果
getPartition返回的值大于等于numReduceTasks或者为负数，运行时会直接报错Illegal partition for record。
2. 案例需求与环境准备
这个案例我选了一个很经典的"手机流量按归属地分区"场景。原因有两个：一是数据直观，手机号前三位大家都懂，归属地也容易理解；二是这个场景完美展示了自定义分区和默认分区的差异，你跑一遍就能看到明显的文件划分效果。
先说清需求。输入数据是CSV格式的文本，每一行记录一个用户某个月的流量使用情况。字段从左到右分别是：手机号、上行流量、下行流量。为了减少代码干扰，我没有放时间戳、基站编号这类无关字段。
输出期望是：按手机号的区号字段（也就是前三位）分区，每个区号生成一个结果文件。结果文件里的每一行是手机号、上行流量、下行流量、总流量。我把测试数据设计成包含多个区号，这样跑完可以清楚看到分出了多少个文件，而且每个文件里只会有一个区号的数据。
换言之，这个案例你的产出不是"所有数据的汇总"，而是"按业务维度切分好的数据集合"。这就是分区能帮你在MapReduce阶段直接完成的额外工作，免去后面再开一轮作业去做分组归档。
2.1 案例场景设定
我准备的模拟数据长这样：
13612345678,100,200 13812345678,300,400 18612345678,200,300 18687654321,150,250 15012345678,500,100 13687654321,120,80这里区号我故意只设计了三种：136、138、186、150。你可以自己随便加。这套数据够小，跑完一眼能看出来每个分区文件里是什么内容，不会像大数据一样只能靠抽样判断。
我假设的归属地映射规则（这里特意用虚构造）：
- 136 对应华东
- 138 对应华东
- 150 对应华南
- 186 对应华北
但你需要注意，真实业务里的区号归属地绝对不是我这样粗略划分的，真正做这种场景需要维护一张完整的号段表。我这里纯粹是为了演示方便。
分区方案是这样：让Map输出的key变成手机号前三位，然后把每个前三位映射到一个分区编号。注意，这里有个很容易踩坑的地方——分区编号不是随便定的，它对应的是ReduceTask的编号或者说是最终输出文件的后缀。后面我细讲。
2.2 环境准备与数据集
这次案例的代码是Java写的，运行环境我用了Hadoop集群。我自己用的是本地起了一个伪分布式环境，具体版本是Hadoop 3.x。如果你用的是Hadoop 2.x，代码基本也能跑，就是包名细节有少许区别，后面会提到。
Java版本我用的JDK 8，虽然高版本Java也能跑，但为了避免Hadoop框架反射部分的兼容性出幺蛾子，建议还是老老实实用JDK 8。Maven工程是必须的，不然依赖管理会让人痛苦。我用的是Maven 3.6以上版本。
数据文件我先放在Linux本地的/tmp/flow/input/flow_data.txt。文件格式是普通的文本，字段用逗号分隔，编码格式用UTF-8，避免中文乱码干扰排查。
环境这里有一个块需要你提前检查：Hadoop的HADOOP_HOME已经配好，bin/hadoop命令能正常执行，yarn的ResourceManager和NodeManager都在运行。如果你用的是伪分布式模式，也请先确认namenode和datanode都活着，不然提交作业会卡在Running job状态很长时间。
注意：网上很多教程喜欢直接在本地IDE里跑MapReduce，不走集群。但分区效果在本地模式下也能看到，只是输出目录路径是
file:///开头，感官上少了点分布式味道。我建议至少跑一次真实的YARN集群模式，这样你才会真正理解"分区"是在分布式环境下发挥作用的。
3. 核心代码实现与逐段拆解
终于进入正题。这个案例的代码分四部分：Mapper、Partitioner、Reducer、Driver。很多人写自定义分区的时候，喜欢把自定义Partitioner跟Driver放在一起甚至写在Reducer后面，虽然也能跑，但逻辑上不够清晰。我建议单独建类，尤其当你的分区规则复杂的时候，独立类更便于维护和测试。
我先给出整体设计，再逐段讲。类结构如下：
FlowPartition ├── FlowPartitionMapper ├── FlowPartitionReducer ├── MyPartitioner └── FlowPartitionDriver一个包下四个类。如果你的需求更复杂，比如一个分区规则里还嵌套判断逻辑，你完全可以再抽一层工具类，但当前案例不需要。
先说Mapper。这个案例的输出key我用手机号前三位字符串，输出value用整个记录的信息。实际上我在value里封装了一个自定义Bean，用来盛放上行流量和下行流量。
具体实现如下：
public class FlowPartitionMapper extends Mapper<LongWritable, Text, Text, FlowBean> { private final Text phonePrefix = new Text(); private final FlowBean flowBean = new FlowBean(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line = value.toString().trim(); if (line.isEmpty() || !line.contains(",")) { return; } String[] fields = line.split(","); if (fields.length < 3) { return; } String phone = fields[0]; long upFlow = Long.parseLong(fields[1]); long downFlow = Long.parseLong(fields[2]); phonePrefix.set(phone.substring(0, 3)); flowBean.set(upFlow, downFlow); context.write(phonePrefix, flowBean); } }注意一个点：phone.substring(0, 3)我取了手机号的前三位，这三位就是区号。为什么要拿这个当key，不拿整个手机号当key？因为分区规则只管前三位，map阶段如果输出完整手机号，分区的时候单独解析key字符串，自然也能做，但会让Partitioner和Mapper耦合，多一道解析流程。直接在Mapper里把key转成前三位，整个后续逻辑都清爽。
FlowBean是一个实现了Writable接口的类，里面有上行流量、下行流量两个字段，以及总流量的getter方法。等一下的Reducer里输出总流量的时候会用到。这个类必须实现write和readFields方法，这就是Hadoop序列化机制的要求，它不让你直接用原生的Java序列化，为了效率和跨语言兼容性，框架自己定义了一套。
这个FlowBean的写法很简单，我把核心部分也列出来：
public class FlowBean implements Writable { private long upFlow; private long downFlow; private long totalFlow; public FlowBean() {} public void set(long upFlow, long downFlow) { this.upFlow = upFlow; this.downFlow = downFlow; this.totalFlow = upFlow + downFlow; } public long getUpFlow() { return upFlow; } public long getDownFlow() { return downFlow; } public long getTotalFlow() { return totalFlow; } @Override public void write(DataOutput out) throws IOException { out.writeLong(upFlow); out.writeLong(downFlow); out.writeLong(totalFlow); } @Override public void readFields(DataInput in) throws IOException { this.upFlow = in.readLong(); this.downFlow = in.readLong(); this.totalFlow = in.readLong(); } @Override public String toString() { return upFlow + "\t" + downFlow + "\t" + totalFlow; } }这里有一个初中级开发容易忽略的点：FlowBean必须有无参构造函数。因为Hadoop在反序列化的时候会通过反射调用无参构造器，你如果只写了带参构造函数，程序会直接抛NoSuchMethodException。这是一种典型的序列化框架限制，跟Java日常开发习惯不太一样。
3.1 自定义Partitioner
这是整个案例的核心。分区器的代码很轻，但逻辑必须清晰。
public class MyPartitioner extends Partitioner<Text, FlowBean> { @Override public int getPartition(Text key, FlowBean value, int numPartitions) { String prefix = key.toString(); switch (prefix) { case "136": case "138": return 0; case "150": return 1; case "186": return 2; default: return 3; } } }解释一下逻辑：我设置了4个分区，编号0、1、2、3。136和138分到0号，150分到1号，186分到2号，其他所有号段分到3号。这意味着我的ReduceTask数量至少要有4个，否则运行时会报"partitions number must be less than reduce task number"之类的错误。
你可能已经感觉到，这种硬编码的switch可维护性很差，真实业务里应该配置一张号段映射表，或者从文件读取映射关系。我这里故意用硬编码，是为了让分区逻辑一目了然，方便你理解机制。实际项目里，推荐用一个Map<String, Integer>来维护号段和分区编号的对应关系，然后加一个默认分区兜底。
还有一件事必须记住：getPartition方法的返回值必须小于传入的numPartitions，且不能为负数。这里的numPartitions其实就是Reducer的数量。如果你不设置setNumReduceTasks，默认是1，那不管你的分区器怎么返回，数据都会写成1个文件，分区器的逻辑相当于完全失效。这是一个特别容易踩的坑，后面我在常见问题里专门讲。
3.2 Reducer实现
Reducer部分属于标准写法，但注意我们这里并没有做任何聚合计算，只是把同一分区的所有记录收集起来依次输出到文件。
public class FlowPartitionReducer extends Reducer<Text, FlowBean, Text, FlowBean> { @Override protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException { for (FlowBean bean : values) { context.write(key, bean); } } }很多教程在这里喜欢做流量求和，把同一个手机号的所有记录聚合成一条。但这个案例的关键是演示分区，不是演示聚合，所以我特意把逻辑降到最低，只做"原样写出"。你看到的结果文件里，同一区号有多少条记录就会写多少行，这样你能更清晰地验证分区效果。
如果将来你要从"按区号分文件"扩展到"按区号聚合流量"，其实就是在Reducer里加一个累加器，把每个key对应的上下行流量分别求和，然后写出一条汇总记录。思路完全一样，关键点只是Reducer里加了几行逻辑而已。
3.3 Driver主程序
Driver部分宁可写复杂一点也要把配置都显式列出来，不要依赖默认值。下面是我的完整Driver实现：
public class FlowPartitionDriver { public static void main(String[] args) throws Exception { if (args.length != 2) { System.err.println("Usage: FlowPartitionDriver <input path> <output path>"); System.exit(-1); } Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "Flow Partition Job"); job.setJarByClass(FlowPartitionDriver.class); job.setMapperClass(FlowPartitionMapper.class); job.setReducerClass(FlowPartitionReducer.class); job.setPartitionerClass(MyPartitioner.class); job.setNumReduceTasks(4); job.setOutputKeyClass(Text.class); job.setOutputValueClass(FlowBean.class); FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }看到setNumReduceTasks(4)这里没？这个数字必须和MyPartitioner里的分区编号最大值对齐。我的分区编号是0到3，所以ReduceTask数量至少是4。如果你设成3，运行时会直接报错；设成5，会产生5个输出文件，最后两个文件是空的或者内容取决于你的数据是否命中其他分区。
3.4 Map输出类型与Reducer输入类型的匹配
还有一点需要特别留意：在Driver里我没有显式设置setMapOutputKeyClass和setMapOutputValueClass。如果Mapper的输出类型和Reducer的输出类型一致，这一步可以省略，因为框架会从Reducer的输入类型推断Mapper的输出类型。但如果Mapper的输出类型和Reducer的输出类型不一样，就必须显式设置setMapOutputKeyClass和setMapOutputValueClass。
我这个案例里Mapper输出<Text, FlowBean>，Reducer输出也是<Text, FlowBean>，类型一致，所以省略掉了。如果你在不同的场景里，Mapper输出的是<Text, IntWritable>，Reducer输出的是<Text, Text>，漏掉setMapOutputKeyClass和setMapOutputValueClass，运行时就会碰到类型转换异常，排查起来比较费劲。
4. 打包、运行与结果验证
代码写完了，接下来就是编译、打包、提交运行。这一套流程说简单也简单，说不简单，打包依赖如果没处理干净，能折腾半天。
我用Maven管理项目，pom里引入hadoop-client依赖。版本号看你的Hadoop环境：
<dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>3.3.4</version> </dependency>打包之前我习惯先跑一次mvn clean package -DskipTests，这样能提前发现编译错误。很多同学遇到跑不出Map 100%结果的情况，其实就是Mapper的代码编译时用了高版本Java语法，而集群的JDK版本却比较低，导致运行时类版本冲突。
4.1 打包与提交命令
打包命令：
mvn clean package -DskipTests生成的jar包路径在target/flow-partition-1.0.jar。提交命令如下：
hadoop jar target/flow-partition-1.0.jar com.example.flow.FlowPartitionDriver /tmp/flow/input/flow_data.txt /tmp/flow/output注意我这里的类名写得完整，com.example.flow.FlowPartitionDriver。如果你在Driver类里没有调用setJarByClass，提交的时候jar参数还是必须带的，这是定位主类的方式之一。我习惯既在代码里setJarByClass，又在命令行指定全类名，双保险。
提交之后你会看到控制台刷出一堆日志。重点看这一段：
Map-Reduce Framework Map input records=6 Map output records=6 Reduce input groups=6 Reduce output records=6 Reduce shuffle bytes=xxx这代表6条输入记录全部被处理，Reducer输出的总记录数也是6条，说明数据没有丢。
4.2 运行结果分析
跑完之后去输出目录看一眼：
hdfs dfs -ls /tmp/flow/output正常情况下你会看到_SUCCESS文件加上多个part-r-xxxxx文件。因为我设置了4个ReduceTask，所以会出现part-r-00000到part-r-00003这4个文件。
然后分别查看每个文件内容：
hdfs dfs -cat /tmp/flow/output/part-r-00000你会看到00000文件里只有136和138号段的数据，00001里只有150号段，00002里只有186号段，00003里是我的测试数据里没有的其他号段，所以是空文件。
这就是分区生效的最直观证据。如果你跑出来4个文件里都有随机混合的多个号段，说明你的MyPartitioner没被框架调用到，或者你的ReduceTask数量没对齐。
还有一个隐藏细节：part-r-00000这个编号和分区编号是对应的。也就是说，你getPartition返回0的数据，最终会进入part-r-00000。这个对应关系别搞混了，调试的时候这是最直接的定位依据。
5. 常见问题与调试实录
这部分我把自己踩过的几个最典型的坑整理出来，这些问题在搜索引擎上天天有人问，也都非常具有代表性。
5.1 分区不生效或数据全到一个区
这个是最常见的。现象是：明明写了自定义Partitioner，也设置了setPartitionerClass，但输出文件只有一个，而且所有数据都在里面。
原因几乎都是同一个：setNumReduceTasks(1)或者根本没设置ReduceTask数量导致默认值是1。框架的默认分区器在你没设置ReduceTask数量的情况下也会工作，但分区编号一律是0，因为一共只有1个分区。自定义分区器可以生效，但是numPartitions这个参数只有1，所以任何返回结果都会被圧到0。
解决办法：在Driver里明确设置job.setNumReduceTasks(4)，并且确认这个数字大于等于你自定义分区逻辑里最大的返回编号加1。
5.2 ReduceTask数量与分区不匹配
分区的返回值必须满足>=0且<numPartitions。如果你返回了4，但ReduceTask只有3个，运行时会报错。报错信息长这样：
Exception in thread "main" java.io.IOException: Illegal partition for record (...)这个错误非常清晰，就是你某个分区编号超过了ReduceTask数量。还有一种隐性问题：你设置了4个ReduceTask，但分区器永远只返回0和1，那最后会跑出4个文件，其中2个是空的。虽然不报错，但不符合预期，浪费了计算资源。
我的建议是分区数量和ReduceTask数量保持相等，并且在自定义分区器里，为所有可能遇到的key类型设置一个默认分区，防止未知数据跑挂作业。
5.3 中文乱码与编码问题
这个案例里的数据虽然全是数字，但如果你在真实业务里把中文传进去，乱码很容易出现。Hadoop默认的Text编码是UTF-8，如果你的输入文件是GBK或者Windows里常出现的ANSI编码，解析出来全是乱码，到了Partitioner里用switch判断字符串，命不中任何case，跑完发现所有数据都落到default分区。
排查技巧也很简单：先不管分区，跑一个最简单的WordCount，看看输出的中文正不正常。如果乱码，先去输入文件那边确认编码，最好在客户端机器上统一转成UTF-8再上传。
5.4 分区与排序的关系
还有一个不少人没意识到的问题：分区本身不保证分区内有序。很多人跑自定义分区的案例，发现输出文件里同一个key的多条记录顺序是乱的，就以为程序写错了。
其实分区只决定"同一类数据进同一个Reducer"，而Reducer内部的数据到达顺序，取决于Map阶段输出的key在分区内的排列，准确说是框架会对每个分区内的key按照键的Comparator去做排序。如果你需要同一个key的多条记录聚在一起容易，因为它们本来就会一起传给Reducer的reduce方法；但如果你要对value再按某一个字段排序，那就需要在排序阶段提前设计好组合键和自定义比较器，这是另一个话题了。
5.5 本地跑通但集群失败
如果你在本地IDE里跑通了此案例，但提交到集群却失败，最可能的原因是Job的setJarByClass没有设置对类。在YARN集群模式下，框架需要通过这个类定位你的jar包。如果你用的是本地模式，这个设置不生效也不会报错，但集群模式下缺失会直接导致无法找到Map类和Reduce类。
另外要注意，如果你的项目用了一堆第三方依赖，不能直接用默认打包方式，否则提交到集群会出现ClassNotFoundException。这种情况推荐用Maven Shade插件把依赖打成一个fat jar，或者你在Driver的main里手动把第三方库加入分布式缓存。我自己的实践是强烈推荐Shade，简单粗暴。
6. 案例的进一步演进思路
案例到这里已经完整跑通了，但我不建议你到这里就停。因为实际业务里，自定义分区很少单独存在，它通常是更大管线里的一环。我总结几个可以自己扩展的方向，做完这些扩展，你对MapReduce的理解会深入很多。
第一个方向是把"流量汇总"加进来。我在Reducer里已经拿到某个手机号前三位下的所有记录，此时只要把上下行流量分别累加，再输出一行汇总结果。这个需求一加，涉及的就是Combiner的取舍和Reducer内部的状态管理，你会更清楚MapReduce的聚合逻辑。
第二个方向是让分区规则配置化。真实的号段映射表可能几千条，不可能写在switch里。你可以把映射关系放到一个文件里，在Driver里用DistributedCache分发给每个Task，在Partitioner里读取这张表来做映射。这一步能让你理解分布式缓存的使用场景，同时模块化程度也更高。
第三个方向是自定义排序实现。假设同一个归属地内部，你想让流量最高的用户排在最前面，那就要让key的结构变成"区号+总流量"的组合，然后用自定义Comparator来指定排序规则，同时分区逻辑还保持在"只看区号"上。这就牵涉到Partitioner、KeyComparator和GroupingComparator三者之间的关系，算是MapReduce阶段很考验功底的玩法。
我在实际摸过几次分区相关的引擎调优之后，最大体会就是：分区看起来只是决定了一个数字，但它贯穿了map输出、shuffle、reduce分配的全链路。尤其在ReduceTask数量比较多的时候，如果分区逻辑不均匀，整个作业的完成时间会被"最长的一个Reducer"拖住，其他节点全部空闲等待。这种情况在数据倾斜的时候尤其明显。
所以说，写分区代码不难，难的是知道你的数据长什么样，以及分区规则会不会导致某个Reducer变成瓶颈。想清楚这一点，你才算真的学会了分区。
