企业官网建设流程全解析

如果你调过MapReduce的并行度，多半遇到过这类怪问题：把mapreduce.job.maps设成1000，实际跑的MapTask还是128个；以为输入文件只有两个，Mapper却拉起了几十个；把文件从10MB加到200MB，集群上Mapper数量纹丝不动。这些现象的背后，绝大多数都不是代码逻辑问题，而是因为你没有理解MapReduce的“分片机制”。

分片（InputSplit）是MapReduce框架对输入数据的逻辑划分，它决定了这个作业要启动多少个MapTask，也直接决定了每份数据会被哪个Mapper读到、从哪个节点读。这部分逻辑在MapReduce源码里并不复杂，核心就集中在FileInputFormat.getSplits()这一个方法里，但它牵扯到HDFS块大小、客户端侧元数据、压缩编解码器、数据本地性等多个概念。这篇文章就从源码的视角，把getSplits()从头到尾拆一遍：goalSize怎么算、splitSize如何被blockSize、minSize、maxSize三方钳制、文件里最后那点“边角料”又是在什么条件下被单独切成一个split的。适合刚准备读Hadoop源码的开发者，也适合那些被数据倾斜、Mapper数量失控折磨过，想从根上把问题看透的人。

1. 分片是什么？先把MapReduce的“输入单位”讲清楚

1.1 分片不等于HDFS块

很多人第一次看到“InputSplit”这个名词时，都会下意识地把它和HDFS块划等号。这是最容易踩的坑。

HDFS块是物理存储概念。NameNode决定一个文件存几个副本、每个副本落在哪些DataNode上，数据在磁盘上是按块存放的。而InputSplit是计算框架的逻辑概念。它不管理数据副本，不参与存储，只是携带了读数据的“说明书”：文件在哪个路径、从哪个偏移开始、需要读多长、可以从哪几台机器上读。

默认情况下，一个split确实会和一块HDFS块大小一致（比如128MB），并且split会尽量对齐到块边界。但这只是因为FileInputFormat在计算splitSize时把blockSize当成了默认上限，并不是说split就是block。你完全可以把一个128MB的文件切出256个1MB的split，也可以把一个10MB的小文件当成一个独立的split，甚至可以用CombineFileInputFormat把100个10MB的小文件装进一个split。这些操作都不会改变HDFS上块的实际布局。

理解了这层差异，后面再看源码就不会觉得别扭了。

1.2 分片在作业生命周期里的位置

一个MapReduce作业从提交到运行，分片计算发生在客户端，而不是在集群的某个节点上。

作业提交时，JobSubmitter会调用InputFormat.getSplits()，框架在这个方法里拿到全部split列表，把split的元数据写入待运行的任务描述中。之后ApplicationMaster收到任务后，才会根据这些split逐个启动MapTask。一个split对应一个MapTask，所以split的数量基本就是MapTask的数量。

这里有个容易混淆的点：split里并不包含真正的数据。它只是一个逻辑描述，实际的数据读取发生在Task启动之后，由RecordReader根据split里的路径、偏移和长度去FileSystem里拉数据。所以split本身非常轻量，这也是为什么MapReduce可以在客户端一次性计算出几万个split并提交，而不会把整个文件内容都拖到客户端。

2. 源码入口：FileInputFormat.getSplits()的完整链路

2.1 为什么分片逻辑都集中在FileInputFormat

MapReduce的InputFormat接口只定义了两个核心方法：getSplits()和createRecordReader()。前者负责切分输入，后者负责从split里读出key/value对。

FileInputFormat是应用最多的实现类，它把“文件列表”逻辑抽出来了，TextInputFormat、SequenceFileInputFormat等具体格式都继承它。也就是说，不管你的输入是文本还是序列文件，只要是普通文件，分片计算的大框架基本一致。你要看的源码入口就是FileInputFormat.getSplits()。

2.2 getSplits()主流程拆解

基于Hadoop 3.x的实现，省略了日志、空创建接口等代码，核心结构如下：

public List<InputSplit> getSplits(JobContext job) throws IOException { // 1. 读取两个全局钳制参数 long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job)); long maxSize = getMaxSplitSize(job); // 2. 列出所有输入文件 List<FileStatus> files = listStatus(job); // 3. 计算总大小与用户期望的map数 long totalSize = 0; for (FileStatus file : files) { if (file.getLen() != 0) { totalSize += file.getLen(); } } long numSplits = job.getConfiguration().getInt(NUM_MAPS, 1); long goalSize = totalSize / Math.max(numSplits, 1L); List<InputSplit> splits = new ArrayList<>(); for (FileStatus file : files) { Path path = file.getPath(); long length = file.getLen(); if (length != 0) { FileSystem fs = path.getFileSystem(job.getConfiguration()); // 4. 获取这个文件对应的BlockLocation信息 BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length); if (isSplitable(job, path)) { long blockSize = file.getBlockSize(); long splitSize = computeSplitSize(goalSize, minSize, maxSize, blockSize); long bytesRemaining = length; // 5. 循环切片，直到剩余量接近一片的大小 while (((double) bytesRemaining) / splitSize > 1.1) { int blkIndex = getBlockIndex(blkLocations, length - bytesRemaining); splits.add(makeSplit(path, length - bytesRemaining, splitSize, blkLocations[blkIndex].getHosts(), blkLocations[blkIndex].getCachedHosts())); bytesRemaining -= splitSize; } // 6. 剩余字节直接作为一个切片 if (bytesRemaining != 0) { int blkIndex = getBlockIndex(blkLocations, length - bytesRemaining); splits.add(makeSplit(path, length - bytesRemaining, bytesRemaining, blkLocations[blkIndex].getHosts(), blkLocations[blkIndex].getCachedHosts())); } } else { // 7. 不可拆分的文件整体作为一个切片 splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts(), blkLocations[0].getCachedHosts())); } } else { // 8. 空文件也保留一个切片，但hosts为空 splits.add(makeSplit(path, 0, length, new String[0])); } } // 9. 把最终split数量写回配置，供后面提交任务时读取 job.getConfiguration().setInt(NUM_MAPS, splits.size()); return splits; }

步骤4里有个很关键的动作：调用fs.getFileBlockLocations(file, 0, length)，一次性拿回这个文件所有HDFS块的Location列表。后面根据偏移找当时的block，再决定split应该带哪些hosts。这也是后面讲数据本地性的基础。

2.3 computeSplitSize：四个变量是怎么博弈的

computeSplitSize是切片大小的核心计算。源码里这个方法并不长，但它背后承载了三个配置参数和一层意图：

protected long computeSplitSize(long goalSize, long minSize, long maxSize, long blockSize) { return Math.max(minSize, Math.min(goalSize, Math.max(maxSize, blockSize))); }

我第一次看到这个公式时，第一反应是“为什么这么绕”。用大白话讲：splitSize先取goalSize和更大的那个上限的较小值，再和minSize比较取较大值。

更直观的理解是下面这组式子：

• 如果没设置maxSize，它的默认值是Long.MAX_VALUE，此时取max(splitSize, blockSize)的结果就是blockSize，所以默认切片上限是块大小，不会切出一个超过128MB的常规split。
• goalSize来自用户期望的MapTask数：totalSize / map个数。如果用户设置mapreduce.job.maps=1600，180GB总输入算出的goalSize就是约112MB，小于块大小，切片就会更贴近用户期望值。
• minSize像一个保底线，默认是Math.max(1, mapreduce.input.fileinputformat.split.minsize)，如果设置了minsize=256MB，那即使blockSize只有128MB，splitSize也至少要取256MB。

如果把源码里的Math.max(maxSize, blockSize)这一步去掉，用效果等价的方式写，就是大多数资料里流传的公式：

splitSize = max(minSize, min(goalSize, blockSize)) if (maxSize != Long.MAX_VALUE) { splitSize = min(splitSize, maxSize) }

也就是说，maxSize只有在显式配置的时候才作为外层的封顶值介入，将已经算好的splitSize压下来。

2.4 参数配置项速查

这里要特别注意，mapreduce.job.maps只是“期望值”，框架会根据最终切分结果把NUM_MAPS重新写回配置。后面第6章会细讲为什么这个参数经常不生效。

3. 文件内切片循环：边角料处理才是源码的精髓

3.1 isSplitable：为什么有些文件会被整块丢给一个Map

在循环里，框架先问isSplitable(job, path)。这个方法本质上是检查压缩格式是否支持“流式分割”。

如果文件是普通文本或者未压缩，Hadoop可以直接从任意字节偏移开始创建一个split，然后在MapTask里从那个位置开始解析出一行完整记录，所以它可拆。如果文件是gzip压缩，GzipCodec没有实现SplittableCompressionCodec接口，框架只能从偏移0开始解压整个流，否则无法在任意split边界处解压出有效数据。因此isSplitable返回false，整个文件会被做成一个split。

这个行为的影响在实战中非常明显：一个20GB的gzip文件，Mapper只有一个，无论集群多大，它都要串行读完。想避免这种情况，建议在数据入仓前就统一用支持可拆分的压缩格式，比如bzip2、lz4、zstd，或者干脆不压缩，让HDFS底层自己做存储管理。

3.2 1.1这个阈值到底在防什么

每次循环切完一片，剩余的字节数可能不够一段完整split。此时源码没有直接再切一片，而是用了一个判定：

while (((double) bytesRemaining) / splitSize > 1.1) { // 继续切一段 splitSize 大小的分片 }

这个1.1意味着：只有当剩余字节大于splitSize的110%时，才值得继续切出一段完整大小的split；否则就把剩余所有字节作为最后一片split。

举例说明会更直观：

• 文件140MB，splitSize按默认128MB计算。剩余量140 / 128 = 1.09，不大于1.1，所以不进入while循环，直接创建出一个140MB的split。这个split比常规的128MB稍大一点，但避免了再启动一个只读12MB数据的Mapper。
• 文件150MB，剩余量150 / 128 = 1.17，大于1.1，先切一个128MB，剩余22MB再单独作为一个split。
• 文件129MB，剩余量129 / 128 = 1.007，不大于1.1，整个文件一个split。

所以这个阈值本质上是“防碎片”：避免因为最后多出几MB数据，就额外启动一个纯浪费的MapTask。代价是允许最后一个split稍微超过splitSize，在极端情况下会达到1.1倍的块大小。

3.3 hosts列表是如何进入split的

每个split在创建时都携带了hosts，这一步对数据本地性至关重要。

在拿到文件的BlockLocation[]之后，如果当前切片的起始偏移是length - bytesRemaining，源码会调用getBlockIndex(blkLocations, start)，找到这个偏移落在第几个HDFS块上，然后把那个块的hosts和cachedHosts传入makeSplit。

因为默认splitSize被控制在blockSize以内，所以大部分情况下一个split只对应一个HDFS块：MapTask如果被调度到该块所在的DataNode上，读取就是本地读。如果split跨越了两个块，它也只能携带起始偏移对应块的主机列表，另一个块的数据仍然可能涉及远程读。这也是为什么不要盲目把splitSize调得比blockSize大太多，越大越容易丢失本地性。

4. 自己动手实现分片：从InputFormat到RecordReader

4.1 为什么需要自定义分片

源码自带的FileInputFormat已经覆盖了绝大多数场景，但总有一些输入不按套路来。常见的自定义需求有几种：

• 希望完全忽略块大小，强制每个文件一个split，方便整体处理一个小数据集。
• 希望按固定行数切分，比如每个Mapper读4096行，而不是按字节数切分。
• 希望把一个目录下的多个小文件合并成一个split。
• 希望从数据库或外部API读数据，根本不走文件系统。

这些都需要实现自定义InputFormat。而自定义InputFormat时，getSplits()和createRecordReader()必须成对出现，因为split只描述了“读哪里”，真正把字节转成key/value的是RecordReader。

4.2 一个最小Demo：固定64MB分片

如果你只是想把切片大小固定成64MB，最简单的做法是继承TextInputFormat，在getSplits()里临时修改配置：

import org.apache.hadoop.mapreduce.InputSplit; import org.apache.hadoop.mapreduce.JobContext; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import java.io.IOException; import java.util.List; public class FixedSizeTextInputFormat extends TextInputFormat { @Override public List<InputSplit> getSplits(JobContext job) throws IOException { long size = 64L * 1024 * 1024; job.getConfiguration().setLong( "mapreduce.input.fileinputformat.split.minsize", size); job.getConfiguration().setLong( "mapreduce.input.fileinputformat.split.maxsize", size); return super.getSplits(job); } }

这里把minSize和maxSize都设成64MB，computeSplitSize最终的结果就会被锁在64MB。使用这个InputFormat会让大文件被切成更多64MB的小split，适合那些处理逻辑非常轻、希望通过增加并行度来压榨资源的场景。

但要注意：这只对可拆分的文件生效。如果输入是gzip压缩文件，isSplitable返回false，整个文件依然会变成一个大split。

4.3 从自定义split到RecordReader

如果只是一味改分片大小，继承现成的TextInputFormat就够了。但如果数据语义变了，比如你要按JSON对象切分、按XML标签切分，就必须实现自定义RecordReader。

RecordReader最重要的初始化方法是：

void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException;

MapTask启动后，框架会把之前算好的split传给RecordReader。自定义RecordReader要根据split.getStart()和split.getLength()打开一个FSDataInputStream，设置读取范围，然后逐条解析出key/value。如果你忽略了split的offset，随便从头开始读，就会在多个Mapper之间重复读数据，甚至读到不属于本split的部分。

在实现的时候，我建议先写一个简单的RecordReader，打印出每轮nextKeyValue()里看到的split offset，再跑一次小文件任务，确认边界是否和预期一致。这个打印习惯能省掉后面大量排查时间。

5. 分片参数调优：三条可以直接复制到生产环境的思路

5.1 调小maxSize来增加Map并行度

当你的任务计算逻辑非常轻，但读入的数据量很大，默认的128MB一个split可能让每个MapTask只跑十几秒，大部分时间都在启动和销毁。为了增加并行度，可以把maxSize调小：

-configuration mapreduce.input.fileinputformat.split.maxsize=33554432

上面这个配置把splitSize压制到32MB。一个10GB的输入文件理论上会切成320个split左右，比默认的80个MapTask多出4倍。加了并行度后，任务整体耗时会明显下降，但也要看到另一面：每一个MapTask都有启动、JIT、调度、shuffle初始化等固定成本，如果任务过小，集群资源反而浪费在框架开销上。一般建议单个MapTask至少运行30秒以上，再考虑继续调小split。

5.2 调大minSize来避免块被切碎

很多新手只知道minSize可以“强制切片不能小于某值”，但在特定场景它更实用：当因为mapreduce.job.maps设置过大导致goalSize非常小，或者某些文件块很碎时，minSize可以把split抬起来，减少MapTask数量。

比如你想让每个split至少256MB，可以设置：

-configuration mapreduce.input.fileinputformat.split.minsize=268435456

这类操作要特别留意一个副作用：当minSize大于blockSize，或goalSize大于blockSize再被minSize放大时，split会跨越多个HDFS块边界，数据本地性可能变差。MapTask可能会被调度到没有该数据块副本的节点，转而通过远程IO读数据。调大minSize之前最好先估算一下文件块布局，别只盯着任务数量。

5.3 CombineFileInputFormat：小文件杀手

小文件问题是MapReduce里的经典痛点。一个目录下有几千个几KB的文件，默认每个文件至少生成一个split，于是产生几千个只有几KB数据的MapTask，调度开销远超数据处理本身。

这种场景用CombineFileInputFormat最合适。它专门把一组小文件打包成一个大split，而不是简单地去改minSize。因为它是在“一个split里包含多个文件”这个维度上做合并，和单个文件内部的切片逻辑完全不同。

使用它也很简单：

import org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat; job.setInputFormatClass(CombineFileInputFormat.class); CombineFileInputFormat.setMaxInputSplitSize(job, 134217728); // 128MB

配置了max split size以后，框架会把相邻的小文件尽量打包到不超过128MB的一个split里。这样几千个小文件最终可能只产生几十个MapTask，实测下来对整体任务时长的改善非常明显。

6. 常见问题和源码排查角度

6.1 为什么改了mapreduce.job.maps不生效

这个现象几乎每个用过MapReduce的人都遇到过。你设置了mapreduce.job.maps=5000，实际Mapper还是只有少量。原因就藏在getSplits()的最后一行：

job.getConfiguration().setInt(NUM_MAPS, splits.size());

用户在客户端配置的mapreduce.job.maps只是参与计算goalSize，会被当作一个期望值，而不是硬性保证。框架真正使用的MapTask数量，是getSplits()算完以后回写的split列表长度。如果你的goalSize因为minSize、maxSize、blockSize这些钳制参数被改回来了，最终split数就不会是5000。

想验证也很简单，用一个只带一个文件、不设任何分片参数的任务，打印Context.getNumReduceTasks()和mapreduce.job.maps，你会看到框架回写后的值。理解了这条回写，就不会再被“不生效”带偏。

6.2 为什么Mapper数量比块数多或少

默认情况下一个大文件的split数约等于文件大小除以块大小，所以很多人直接拿“块数”预测Mapper数。但这个说法只在文件恰好等于块大小的整数倍时才成立。

如果文件最后剩余量在1.1倍splitSize以内，会直接作为一整段split，导致split数比“文件大小/块大小”的数学结果少一。如果文件不可拆分，那么无论文件有多少个块，都只有一个split。如果文件较小但数量很多，那么每个文件一个split，Mapper数直接和文件数挂钩。因此，精确计算split数最可靠的办法是看客户端日志里的Splits=信息，它会把每个split的offset和长度都打出来。

6.3 一个gzip文件拖垮整个作业

数据倾斜的排查里，最容易被忽略的就是压缩格式导致的单MapTask瓶颈。一个几十GB的gzip文件会让isSplitable返回false，整个task在集群里只能单节点顺序读。从split源码的角度看，它甚至不需要进入切片while循环，直接执行makeSplit(path, 0, length, hosts)。

这种问题最彻底的解法是换用支持可拆分的压缩编解码器，或者在写入数据时换成parquet/orc这类列存格式。如果临时要跑，只能硬扛单个Mapper的物理读上限。

6.4 源码阅读路径建议

如果读完这篇还想自己翻源码验证，我推荐按下面顺序看：

• FileInputFormat.getSplits()：整个分片主流程。
• FileInputFormat.computeSplitSize()：确认splitSize的三方博弈细节。
• SplitLocationRetriever.getBlockIndex()：理解偏移和块索引的关系。
• LineRecordReader.initialize()：看看split的offset和length是怎么被真正用于读取文件的。

这四段源码是一条完整链路：分片计算到读取执行。我自己第一次完整读完时，最大的感受是“终于明白为什么MapTask数量和直觉差那么远了”。这里面的逻辑并不复杂，但每一行都踩在分布式计算的资源分配和本地性取舍上。如果你也在调并行度，强烈建议不要只背参数，花点时间把这些源码支线走一遍，收益会远超预期。