如果你已经用Hadoop跑了一两年MapReduce，大概率经历过这种场景：作业提交上去，任务列表里突然冒出来几百个Map任务，或者明明数据量很大却只有两三个Map任务，你挠头半天也不知道该怎么调。这个问题的答案，藏在MapReduce的分片机制里，具体说，就是在FileInputFormat.getSplits()这段源码里。分片机制决定了输入数据怎么被切分、Map任务并行度怎么定、数据本地性怎么选，是MapReduce性能调优绕不开的一环。这篇博文我就以源码为主线，把分片机制整个拆一遍，从接口设计到计算逻辑再到常见坑，给正在学Hadoop或遇到作业性能瓶颈的朋友一份能直接落地的参考。
1. 分片机制到底在解决什么问题
1.1 什么是分片：Map任务的输入单元
先明确一个概念：分片，源码里叫InputSplit，它不是HDFS里的数据块，而是MapReduce在逻辑层面对输入数据的划分。一个分片就是一组记录的集合，MapReduce框架会给每个分片分配一个Map任务。换句话说，分片数量直接决定了Map任务数。
我记得最开始的困惑就在这里：为什么不能直接拿文件来当单位？因为文件大小差异太极端了。一个目录下可能有几百个几KB的小文件，也可能有一个几百GB的大文件。如果直接按文件分配Map任务，小文件那批会一下启动几百个Map，每个只处理几秒钟，大部分时间都耗在任务启动和资源调度上；而那个大文件只有一个Map任务，要跑几个小时才能处理完，集群资源大部分时间被它独占。
分片机制就是为了解决这个问题：它把文件按照规则切成一个个逻辑片，每个片对应一个Map任务，这样粒度可控、负载均匀，并行度也能按需调整。
1.2 分片和HDFS块的正确关系
块是物理概念，分片是逻辑概念，这是最容易混淆的地方。
HDFS块是存储层的数据切割单位，默认128MB，分布在不同的DataNode上。分片是计算层的数据切割单位，它需要找到自己的数据在哪、从哪个偏移量开始读、读多长。框架在生成分片时会记录数据所在的块位置，这样调度器就能优先把Map任务分配到数据本地的节点上，减少网络传输，这就是数据本地性。
默认情况下，分片大小和块大小是一致的——都是128MB左右。但这不是必须的。你完全可以把分片设成256MB、64MB或者1GB，块不会因此改变，只是你的一个分片可能跨越多个物理块，数据本地性就会打折扣。这个关系后面看源码时会更清楚。
1.3 为什么不能直接拿块当分片
你可能会问：既然HDFS已经按块把文件切开了，为什么MapReduce不直接用块做分片？
原因有两个。第一，块是面向存储冗余设计的，它不关心怎么切分逻辑记录。拿文本文件举例，一个块在任意位置断开，下一行的前半段在这个块、后半段在下一个块，如果直接把块交给Map任务处理，就会有一堆行被读碎，还需要跨节点拼接，复杂且低效。
第二，分片需要携带启动Map任务所需的位置信息。一个块可以知道它在哪个DataNode，但一个Map任务其实要读取多个块的数据（当分片大于块大小时），这时候就需要把多个块的位置信息合并到一个分片里。块自己做不到这件事。所以框架在InputSplit中专门封装了getLocations()方法，用来返回分片数据所在的所有节点列表。
2. 核心源码阅读路径：从FileInputFormat开始
2.1 InputFormat的三个核心方法
分片机制最顶层的接口是InputFormat，它只有三个方法：
public interface InputFormat<K, V> { // 切分输入数据，生成一系列InputSplit List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException; // 为每个分片创建一个RecordReader，用于逐条读取记录 RecordReader<K, V> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException; }getSplits()负责切分，createRecordReader()负责把分片里的数据转换成(key, value)记录。
平时我们最常用的TextInputFormat，它继承自FileInputFormat，而FileInputFormat实现了getSplits()的核心逻辑。所以看分片机制，主要就是看FileInputFormat的这段实现。
2.2 getSplits()源码逐行拆解
下面这段代码是FileInputFormat.getSplits()的精简版，保留了核心逻辑。每个版本细节略有差异，但骨架基本一致：
public List<InputSplit> getSplits(JobContext job) throws IOException { // 读取两个关键参数 long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job)); long maxSize = getMaxSplitSize(job); List<InputSplit> splits = new ArrayList<InputSplit>(); List<FileStatus> files = listStatus(job); for (FileStatus file : files) { Path path = file.getPath(); long length = file.getLen(); if (length != 0) { FileSystem fs = path.getFileSystem(job.getConfiguration()); // 获取文件所有块的位置信息，这是数据本地性的基础 BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length); long blockSize = file.getBlockSize(); long splitSize = computeSplitSize(blockSize, minSize, maxSize); long bytesRemaining = length; // SPLIT_SLOP = 1.1，剩余数据不足分片大小的1.1倍时就不再切分 while (((double) bytesRemaining) / splitSize > SPLIT_SLOP) { int blkIndex = getBlockIndex(blkLocations, length - bytesRemaining); splits.add(makeSplit(path, length - bytesRemaining, splitSize, blkLocations[blkIndex].getHosts(), blkLocations[blkIndex].getCachedHosts())); bytesRemaining -= splitSize; } // 处理最后剩余的不足一个分片的数据 if (bytesRemaining != 0) { int blkIndex = getBlockIndex(blkLocations, length - bytesRemaining); splits.add(makeSplit(path, length - bytesRemaining, bytesRemaining, blkLocations[blkIndex].getHosts(), blkLocations[blkIndex].getCachedHosts())); } } else { // 空文件也生成一个长度为0的分片 splits.add(makeSplit(path, 0, length, new String[0])); } } return splits; }这段代码看着简单，信息量其实很大。
先说getFormatMinSplitSize()。它在FileInputFormat里默认是1字节，但有些子类会重写，比如CombineFileInputFormat就会返回一个较大的值，把很多小文件合并成一个分片。
再看minSize的取值：Math.max(getFormatMinSplitSize(), getMinSplitSize(job))。getMinSplitSize(job)读取的是配置项mapreduce.input.fileinputformat.split.minsize，默认是1。所以一般情况下，minSize就是1字节，对分片计算没有实际影响。除非你显式设置这个参数。
maxSize读取的是mapreduce.input.fileinputformat.split.maxsize，默认是Long.MAX_VALUE，也就是不限制上限。一般情况下也不用管。
2.3 分片大小计算公式
真正决定分片大小的是computeSplitSize()：
protected long computeSplitSize(long blockSize, long minSize, long maxSize) { return Math.max(minSize, Math.min(maxSize, blockSize)); }这个公式的含义：分片大小取minSize和maxSize之间夹住blockSize的中间值。
默认情况下，minSize=1，maxSize=Long.MAX_VALUE，所以结果就是blockSize，也就是128MB。这也解释了为什么默认分片大小等于块大小。
如果我把maxSize设成64MB，公式变成Math.max(1, Math.min(64MB, 128MB))，结果是64MB。每个分片变小，分片数量变多，Map任务数变多。
如果我把minSize设成256MB，公式变成Math.max(256MB, Math.min(Long.MAX_VALUE, 128MB))，结果是256MB。每个分片变大，Map任务数变少。
这里有一个隐藏的细节：分片大小是逻辑层面的，它不要求对齐物理块边界。一个256MB的分片会跨两个甚至多个HDFS块，getBlockIndex()会根据当前分片的起始偏移量找到它所在的那一块，然后把这一块的节点列表作为分片的locations。这意味着当分片比块大时，分片只能把数据量最大那一块的位置作为本地性参考，数据本地性会下降。
还有一个细节是SPLIT_SLOP。它是1.1，也就是当剩余数据量除以分片大小小于等于1.1时，不再继续切分，把剩余数据全部作为一个分片。为什么这么设计？因为如果严格按照bytesRemaining / splitSize来切，一个128MB的文件会被切成1个128MB的分片；但如果文件是130MB，就会切成两个分片：一个128MB、一个2MB。这个2MB的Map任务启动开销和128MB的完全一样，但处理的数据量差了60多倍，非常浪费。SPLIT_SLOP就是避免生成这种极端小分片用的，它允许最后一个分片稍微超过标准大小，只要不超过1.1倍就容忍。
2.4 makeSplit：分片里到底装了什么
makeSplit()最终生成一个FileSplit对象，它的几个核心字段是：
private Path file; // 文件路径 private long start; // 分片在文件中的起始偏移量 private long length; // 分片长度 private String[] hosts; // 数据所在节点可以看到，一个分片并不复制数据本身，它只是一个描述性的指针：告诉框架“去哪个文件的哪个偏移量开始读、读多少、数据在哪个节点上”。真正读数据是后面RecordReader干的活。
getBlockIndex()的算法是：遍历文件的所有块位置，找到偏移量offset落在哪一块的[start, start+length)区间内，返回该块的索引。这个偏移量就是当前分片的起始位置。为什么选块的节点列表而不合并所有相关块的节点？因为一个分片跨多个块时，合并所有节点会把本地性列表拉得太长，调度器会无所适从，所以只取起始块的位置，这是性能与准确性的权衡。
3. 分片数据到Mapper的桥接
3.1 RecordReader如何消费分片
分片规划好了，下一步就是读数据。FileInputFormat在生成分片后，MapTask会调用createRecordReader()为分片创建对应的RecordReader，逐条读取记录。
public RecordReader<K, V> createRecordReader(InputSplit split, TaskAttemptContext context) { return new LineRecordReader(); }LineRecordReader的初始化方法很关键：
public void initialize(InputSplit genericSplit, TaskAttemptContext context) throws IOException { FileSplit split = (FileSplit) genericSplit; // 打开文件输入流，定位到分片起点 FSDataInputStream fileIn = fs.open(split.getPath()); fileIn.seek(split.getStart()); // 读取第一行 ... }这里有个容易被忽略的点：LineRecordReader在起点处读取时，会判断自己读取的第一个字符是不是换行符。如果是，说明起点正好在行尾，那就跳过这个换行符，从下一行开始。为什么要这么做？因为分片边界不一定落在行的开始位置，可能是从某行的中间开始读。为了保证每一行都完整读到，LineRecordReader会从分片起点开始，如果发现起点不在行首，它会把这一行“吞掉”，但不会输出给Mapper。
这带来的一个结果是：一个Map任务实际读取的数据量可能比分片长度稍微多一点——它会多读一段到下一个换行符为止的内容。这就是为什么你经常能看到某些Map任务的输入字节数比日志里的split大小大一点点。
3.2 TextInputFormat的文本处理细节
TextInputFormat默认把每行的偏移量作为Key，把那行文本作为Value。它依赖LineRecordReader逐行读取。
如果分片边界正好切在一行的中间，这一行会被前面的Map任务漏掉吗？不会。回到上面的逻辑：分片B的起始位置在线中间，它读取时会先把前半行读完并丢弃，然后从完整的一行开始输出。这样分片A负责前半行的记录（它读到行尾，然后输出），分片B丢弃同一行的前半部分，再从下一行开始。整条记录只被输出一次，不重复也不丢失。
这个机制看起来简单，但涉及跨节点读数据的事故排查时，理解它很有用。比如你发现某个Map任务明明只处理了64MB数据，实际上却读了64MB加一行很长的文本，这不能算bug，是框架刻意为之。
3.3 压缩文件与isSplitable
分片机制有个重要前提：文件是可切的。FileInputFormat里有一个方法：
protected boolean isSplitable(JobContext context, Path filename) { return true; }TextInputFormat重写了这个方法：
@Override protected boolean isSplitable(JobContext context, Path filename) { final CompressionCodec codec = new CompressionCodecFactory(context.getConfiguration()).getCodec(filename); if (null == codec) { return true; } return codec instanceof SplittableCompressionCodec; }什么意思？如果你的文件是gzip压缩格式，GzipCodec不是SplittableCompressionCodec，那么isSplitable返回false，整个gzip文件只会生成一个分片，不管这个文件有多大。因为gzip是串行压缩格式，解压必须从头开始，前面数据没解出来，后面的数据解不了。
如果文件是去掉加的bzip2格式，BZip2Codec实现了SplittableCompressionCodec，那么它依然可以切分，只是每个分片启动时要多做一些块同步工作。
这个细节在真实调优里非常常见：一个500GB的gzip文件，无论你怎么调split.maxsize，它都只有一个Map任务，性能上不去的根本原因在压缩格式，不在分片参数。
3.4 自定义InputFormat实例
分片机制不是只能接受默认值，很多场景需要自定义。我记得大约一个大项目需要把多个文件组合成一个分片处理，默认的TextInputFormat会导致每个文件生成一个分片，无法合并计算。当时用的方案是继承CombineFileInputFormat：
public class MyCombineInputFormat extends CombineFileInputFormat<LongWritable, Text> { public MyCombineInputFormat() { super(); // 每个分片最大64MB，控制Map任务数量 setMaxSplitSize(64 * 1024 * 1024); } @Override public RecordReader<LongWritable, Text> createRecordReader(InputSplit split, TaskAttemptContext context) { return new CombineFileRecordReader<>((CombineFileSplit) split, context, MyRecordReader.class); } }CombineFileInputFormat的特别之处在于：它会把同一个目录下多个小文件拼接成逻辑分片，每个分片里包含多个文件片段。这样Map任务只要处理少量几个大分片，而不需要为每个小文件单独启动一个Map，能极大地缓解小文件场景下的资源浪费。
要实现一个完全自定义的InputFormat，核心就是三件事：
- 继承
InputFormat或FileInputFormat - 实现
createRecordReader()，返回自己的RecordReader - 在
RecordReader.initialize()里定义好从FileSplit读取数据的逻辑
分片切分的逻辑只要你不重写getSplits()，它仍然沿用FileInputFormat的默认切分规则。
4. 实操：调整分片参数的几种方式
4.1 两大核心参数
调整分片大小，最终就是调整那两个配置项：
| 配置项 | 默认值 | 作用 |
|---|---|---|
mapreduce.input.fileinputformat.split.minsize | 1 | 分片最小字节数，设大可以让分片变大、Map数变少 |
mapreduce.input.fileinputformat.split.maxsize | Long.MAX_VALUE | 分片最大字节数，设小可以让分片变小、Map数变多 |
在代码里设置的方式：
Configuration conf = new Configuration(); // 让每个分片至少256MB conf.set("mapreduce.input.fileinputformat.split.minsize", "268435456"); // 让每个分片最大64MB conf.set("mapreduce.input.fileinputformat.split.maxsize", "67108864"); Job job = Job.getInstance(conf); job.setInputFormatClass(TextInputFormat.class);或者在提交脚本里直接写：
hadoop jar myjob.jar MyJob \ -D mapreduce.input.fileinputformat.split.minsize=268435456 \ -D mapreduce.input.fileinputformat.split.maxsize=67108864这里有几个字节数的换算要点：128MB是128乘1024乘1024，也就是134217728字节。很多人图省事直接写128，结果是128字节，整个分片机制直接失效，Map任务数量疯涨，排查半天找不到原因。
4.2 三个典型场景
场景一：大量小文件，每个文件只有几百KB，但数量有10万个。
默认情况下每个文件生成至少一个分片，10万个Map任务，每个Map启动就要消耗几秒，处理数据只有几百KB，集群资源全部浪费在调度上。这时候有两种选择：一是用CombineFileInputFormat把文件合并成分片；二是把split.minsize设置成128MB，让FileInputFormat把多个文件合并到同一个分片里。注意后者的合并逻辑和CombineFileInputFormat不完全一样，但对减少Map数量同样有效。
场景二：单个文件300GB，默认分片128MB，会产生大约2344个Map任务。
如果集群只有20个核心，每个Map跑5分钟，2344个Map全挤在一起反而可能因为资源等待变慢。这种情况可以把split.maxsize调大，比如设成1GB，让分片变少，Map任务数降到300左右，每个任务处理时间变长，整体调度开销反而更小。
场景三：处理速度要求高，希望尽量充分利用集群所有节点。
可以减小split.maxsize到32MB，让分片变多、Map并行度变高。但一定要权衡：分片太小的代价是shuffle阶段的网络和排序开销会显著上升，因为Map输出的小结果集数量更多。我见过把分片调到16MB的作业，Map阶段确实快了不少，reduce阶段却明显变慢，综合下来毫无收益。
4.3 分片数量如何预估
在实际调优时，可以通过下面的公式快速估算Map任务数：
当所有文件大小相近且大于分片大小时： Map数 = 总数据量 / 分片大小（向上取整）
当存在大量小文件时，哪怕设置较大的minsize，每个文件也至少会生成一个分片，Map数不会低于文件数。这个规律在getSplits()源码里看得很清楚：空文件生成一个长度为0的分片，非空文件即使只有1KB，也会进入循环并生成一个分片。
预估Map数之后对比集群的并发度（比如YARN分配的容器数），就能知道瓶颈在资源等待还是数据倾斜。
5. 常见问题与排查技巧实录
5.1 问题速查表
| 现象 | 可能原因 | 解决方案 |
|---|---|---|
| Map任务数量异常多，几百上千个 | 输入目录小文件过多，每个文件至少一个分片 | 用CombineFileInputFormat或调大split.minsize |
| Map任务数很少，但单个Map处理时间极长 | 分片过大、文件不可切分、数据倾斜 | 调小split.maxsize；检查压缩格式是否可切分 |
| 某个Map任务读取数据量超过分片大小 | LineRecordReader为了补全行，会多读一段到行尾 | 正常现象，无需处理 |
调整split.minsize/maxsize后Map数量没变化 | 文件是gzip等不可切分压缩格式，整个文件只生成一个分片 | 换可切分压缩格式，或解压后处理 |
| 多个Map任务访问同一份数据，网络开销大 | 分片大小与块大小不匹配，分片跨越多个块，本地性下降 | 尽量保持分片大小等于块大小 |
| Map任务数接近文件数，而非总数据量/分片大小 | 有大量小文件在分片循环之前就各自生成了一个分片 | 合并小文件，或使用CombineFileInputFormat |
5.2 如何确认当前作业的split分布
排查分片问题时，不要靠猜。你可以直接在作业的计数器里看两个指标：
FileInputFormatCounter.BYTES_READ：实际读取的字节数FileInputFormatCounter.BYTES_WRITTEN：Map输出字节数（通常为0）
更直接的方式是临时写一个打印分片的工具类，把getSplits()的输出打印出来：
Job job = Job.getInstance(conf); TextInputFormat inputFormat = new TextInputFormat(); List<InputSplit> splits = inputFormat.getSplits(job); for (InputSplit split : splits) { FileSplit fileSplit = (FileSplit) split; System.out.println("file=" + fileSplit.getPath() + ", start=" + fileSplit.getStart() + ", length=" + fileSplit.getLength() + ", hosts=" + Arrays.toString(fileSplit.getLocations())); }这个工具在验证小文件场景、或者确定一个超大文件到底被切成多少个分片时，比任何日志都直观。
5.3 数据倾斜：分片均匀不等于负载均匀
最后说一个最容易踩的坑：分片机制只能保证输入数据的字节数均匀，不能保证每条记录的处理成本均匀。
比如你的业务是按用户聚合作业，80%的用户都集中在一个分片里，那这个Map任务要处理的数据和别的分片差不多，但计算时间会远超其他任务。这种情况在分片层面看不出任何异常，所有分片都是128MB，漂漂亮亮的。
解决办法有两种思路：一种是在Map阶段想方设法把热点key打散，比如加随机盐、两次聚合；另一种是让最终Reducer的输入均匀。前者通常在业务层处理，分片机制本身无能为力。所以当出现“明明分片数合理，作业还是慢”的场景，先把数据分布和key分布检查一遍，再回头看分片参数。
6. 一个调优实例的复盘
拿我之前遇到的一个真实案例收尾。某离线统计分析作业，每天处理大约500GB文本数据，输入目录下混着几千个小文件和十几个大文件。默认参数跑下来，Map任务数一度超过4000，YARN队列被占满，其他作业全被堵住。
排查过程分三步。先打印getSplits()的结果，确认超过九成的分片都是小文件产生的：每个几KB到几十KB的文件都对应一个分片。再检查大文件的分片是否正常，结果都是128MB的标准切分。最后看压缩格式，确认输入文件都是未压缩的普通文本。
处理方式是给作业设置CombineFileInputFormat，并把每个分片的最大大小控制在256MB。改完以后Map任务数降到不到500，每个Map处理的数据量都在合理范围，集群队列也不再被挤爆。相比之下，单纯把split.minsize调大也能减少Map数量，但CombineFileInputFormat在处理小文件时对文件列表的合并更高效，更不容易产生边界残留。
这个案例里最值得记的一点是：分片机制不是越调越大越好，也不是越小越并行，而是要和你的数据规模、集群内存、队列容量匹配。了解了getSplits()的源码逻辑之后，你再看到Map任务数量异常，第一反应就不应该是“改个参数试试”，而是先想清楚这里的三个变量——minSize、maxSize、blockSize——到底哪个在起作用，文件能不能切分，小文件占了多少比重。
我个人在踩过几次坑之后，养成了一个习惯：接手任何MapReduce作业，第一件事不是看业务逻辑，而是先打印一份getSplits()的结果，把文件清单、分片大小、Map数量三者铺开放在一起看。分片这层理顺了，后续的Map并行度、本地性、shuffle开销才有讨论的基础。你也可以试试这个方法，说不定能少走不少弯路。
