好的，各位观众老爷们，欢迎来到今天的MapReduce奇妙夜！今晚我们要聊的，是MapReduce中一个非常关键，但又常常被大家忽略的幕后英雄——InputFormat的Splits逻辑，也就是传说中的“数据切分”！
想象一下，你面前有一座金矿，哦不，是TB级别的数据金矿！如果你想快速挖到里面的宝藏（有价值的信息），一股脑地把整个矿山都搬回家，那肯定是不现实的。正确的做法是什么呢？当然是把矿山分成一块一块的区域，然后分给不同的矿工（Mapper）同时挖掘，这样效率才能蹭蹭往上涨嘛！而这个“分矿”的过程，就是InputFormat的Splits逻辑在做的事情。
一、InputFormat：数据入口的守门员
首先，我们得认识一下InputFormat。这家伙就像是MapReduce程序的数据入口的守门员，负责以下几项重要的工作：
- 验证输入数据的格式： 确保输入数据是MapReduce可以处理的类型。就好比你去参加一个晚宴，总得穿上合适的礼服才能进去吧？InputFormat会检查你的数据“礼服”是否合身。
- 数据切分 (Splitting)： 这是我们今天的主角！将输入数据分割成多个小的逻辑分片 (Splits)，每个Split会被一个Mapper处理。
- 提供RecordReader： RecordReader负责从Split中读取数据，并将数据转换成键值对 (Key-Value Pairs) 的形式，供Mapper使用。你可以把RecordReader想象成一个翻译官，把矿石翻译成矿工能理解的语言。
二、Splits：让数据“瘦身”的魔法
现在，让我们聚焦到Splits！Splits是InputFormat最重要的工作之一，它的目标是将巨大的输入数据分割成多个小的、可并行处理的逻辑单元。 为什么要这么做呢？原因很简单：
- 并行处理： MapReduce的核心思想就是并行处理。将数据分割成多个Splits后，就可以分配给多个Mapper同时处理，大大提高了处理速度。
- 负载均衡： 良好的Splits策略可以确保每个Mapper处理的数据量大致相同，避免出现某些Mapper“吃不饱”，某些Mapper“撑死”的情况，从而实现更好的负载均衡。
- 容错性： 如果某个Mapper处理失败，只需要重新处理对应的Split即可，不会影响其他Mapper的工作。
三、Splits的切割艺术：技术与智慧的交织
那么，InputFormat是如何切割数据的呢？这可不是随便乱切一刀的事情，而是需要根据不同的数据格式和需求，采用不同的切割策略。常见的切割策略包括：
- 基于文件大小的切割： 这是最简单的一种策略。将文件按照固定的大小（例如128MB，HDFS默认的Block大小）进行分割。这种策略适用于大多数情况，但可能会导致某些Split包含不完整的记录。
- 基于记录边界的切割： 这种策略会根据记录的边界（例如换行符）进行分割，确保每个Split包含完整的记录。这种策略适用于文本文件等有明显记录边界的数据。
- 自定义切割策略： 对于一些特殊的数据格式，可能需要自定义切割策略。例如，对于XML文件，可以根据XML的标签进行分割，确保每个Split包含完整的XML文档。
让我们用一个表格来总结一下：
| 切割策略 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 基于文件大小的切割 | 大多数情况，特别是二进制文件 | 简单高效 | 可能导致Split包含不完整的记录 |
| 基于记录边界的切割 | 文本文件等有明显记录边界的数据 | 确保每个Split包含完整的记录 | 不适用于没有明显记录边界的数据 |
| 自定义切割策略 | 特殊的数据格式，例如XML文件、JSON文件等 | 可以根据数据的特点进行精确的切割，提高处理效率 | 实现较为复杂，需要编写额外的代码 |
举个栗子🌰：
假设我们有一个巨大的文本文件，里面存储了用户日志，每行代表一条日志记录。如果使用基于文件大小的切割策略，可能会出现这样的情况：
Split 1:
2023-10-27 10:00:00 UserA logged in
2023-10-27 10:00:05 UserB logged out
2023-10-27 10:00:10 UserC
Split 2:
logged in
2023-10-27 10:00:15 UserD logged in
2023-10-27 10:00:20 UserE logged out可以看到，Split 2包含了一条不完整的日志记录 "logged in"。 这可能会导致Mapper在处理Split 2时出错。
如果使用基于记录边界的切割策略，就可以避免这个问题：
Split 1:
2023-10-27 10:00:00 UserA logged in
2023-10-27 10:00:05 UserB logged out
2023-10-27 10:00:10 UserC logged in
Split 2:
2023-10-27 10:00:15 UserD logged in
2023-10-27 10:00:20 UserE logged out每个Split都包含了完整的日志记录，Mapper可以正常处理。
四、深入源码：窥探Splits的秘密花园
光说不练假把式，让我们深入到InputFormat的源码中，看看Splits是如何被切割出来的。以TextInputFormat为例，这是最常用的处理文本文件的InputFormat。
public class TextInputFormat extends FileInputFormat<LongWritable, Text> {
@Override
public RecordReader<LongWritable, Text> createRecordReader(InputSplit split,
TaskAttemptContext context) {
String delimiter = context.getConfiguration().get("textinputformat.record.delimiter");
RecordReader<LongWritable, Text> rr;
if (null != delimiter) {
rr = new DelimitedLineRecordReader();
} else {
rr = new LineRecordReader();
}
return rr;
}
@Override
public List<InputSplit> getSplits(JobContext job) throws IOException {
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);
if (isSplitable(job, path)) {
long blockSize = file.getBlockSize();
long splitSize = computeSplitSize(job,
minSplitSize,
blockSize);
long bytesRemaining = length;
while (((double) bytesRemaining) / splitSize > SPLIT_SLOP) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(new FileSplit(path, length-bytesRemaining, splitSize,
blkLocations[blkIndex].getHosts()));
bytesRemaining -= splitSize;
}
if (bytesRemaining != 0) {
int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining,
blkLocations[blkIndex].getHosts()));
}
} else { // not splitable
splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));
}
} else {
//Create empty splits for zero length files
splits.add(new FileSplit(path, 0, length, new String[0]));
}
}
// Save the number of input files split into the configuration
job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
LOG.debug("Total # of splits: " + splits.size());
return splits;
}
}这段代码的核心逻辑在getSplits方法中，它做了以下几件事情：
- 获取所有输入文件： 通过
listStatus方法获取所有输入文件的信息。 - 遍历每个文件： 对每个文件进行处理。
- 判断文件是否可分割： 通过
isSplitable方法判断文件是否可以分割。对于一些压缩文件（例如Gzip文件），通常是不可分割的。 - 计算Split大小： 通过
computeSplitSize方法计算Split的大小。Split的大小受到mapreduce.input.fileinputformat.split.minsize和mapreduce.input.fileinputformat.split.maxsize等参数的影响。 - 切割文件： 根据Split的大小，将文件切割成多个Splits。
- 创建FileSplit对象： 为每个Split创建一个
FileSplit对象，其中包含了Split的文件路径、起始位置、长度以及所在的Block的位置信息。 - 返回Splits列表： 将所有Splits添加到一个列表中，并返回该列表。
五、Splits与HDFS Block：剪不断理还乱的关系
你可能会问，Splits的大小和HDFS的Block大小有什么关系呢？ 这是一个非常重要的问题！
- 理想情况下，一个Split应该包含一个或多个完整的HDFS Block。 这样可以最大程度地利用HDFS的本地性优势，减少数据传输的开销。
- InputFormat会尽量将Split放置在存储该Split对应数据Block的DataNode上。 这样Mapper就可以直接从本地读取数据，而不需要通过网络传输数据，大大提高了处理效率。
用一个生动的比喻来说：
HDFS Block就像是一个个集装箱，里面装着数据。而Split就像是一个订单，指定了需要从哪些集装箱中提取哪些货物。 InputFormat的任务就是尽可能地让每个订单都对应到同一个DataNode上的集装箱，这样就可以减少运输成本，提高效率。
六、自定义InputFormat：打造专属的数据入口
虽然Hadoop提供了多种InputFormat，例如TextInputFormat、SequenceFileInputFormat、AvroInputFormat等，但有时候我们需要处理一些特殊的数据格式，这时就需要自定义InputFormat。
自定义InputFormat的步骤如下：
- 继承
InputFormat类或其子类： 例如，可以继承FileInputFormat类。 - 实现
createRecordReader方法： 该方法负责创建RecordReader对象，用于从Split中读取数据。 - 实现
getSplits方法： 该方法负责将输入数据分割成多个Splits。
自定义InputFormat的关键在于如何实现getSplits方法。 你需要根据数据的特点，设计合适的切割策略，并编写相应的代码。
七、Splits优化：让MapReduce飞起来
Splits的切割策略对MapReduce程序的性能有着重要的影响。 如何优化Splits，让MapReduce飞起来呢？
- 合理设置Split的大小： Split的大小应该根据集群的规模、数据的特点以及Mapper的处理能力进行调整。 过小的Split会导致Mapper频繁启动和关闭，增加开销。 过大的Split会导致Mapper处理时间过长，降低并行度。
- 选择合适的切割策略： 根据数据的格式和特点，选择合适的切割策略。 例如，对于文本文件，应该使用基于记录边界的切割策略。
- 使用CombineFileInputFormat： 对于小文件较多的情况，可以使用
CombineFileInputFormat将多个小文件合并成一个Split，减少Mapper的数量，提高效率。
八、总结：Splits，MapReduce的基石
今天我们深入探讨了MapReduce中InputFormat的Splits逻辑。 Splitting是将大数据分解为可并行处理的小块数据的关键步骤，直接影响MapReduce作业的效率和性能。 通过理解不同的切割策略、深入源码以及掌握优化技巧，我们可以更好地利用MapReduce处理海量数据，挖掘出更多的宝藏！
希望今天的分享能让你对MapReduce的Splits逻辑有更深入的理解。 记住，掌握Splits，就掌握了MapReduce的基石！ 下次再见！ 👋