Hadoop FileInputFormat实现原理及源码分析

FileInputFormat（org.apache.hadoop.mapreduce.lib.input.FileInputFormat）是专门针对文件类型的数据源而设计的，也是一个抽象类，它提供两方面的作用：

 

（1）定义Job输入文件的静态方法；

（2）为输入文件形成切片的通用实现；

 

至于如何将切片中的数据转换为一条条的“记录”则根据文件类型的不同交由具体的子类负责实现。

 

 

FileInputFormat提供了四个静态方法用于定义Job的输入文件路径：

 

public static void addInputPath(Job job, Path path)

public static void addInputPaths(Job job, String commaSeparatedPaths)

public static void setInputPaths(Job job, Path... inputPaths)

public static void setInputPaths(Job job, String commaSeparatedPaths)

 

addInputPath()、addInputPaths()用于“添加”一个（批）输入路径，可以重复被调用，源码如下：

 

 

addInputPath的每一次调用都会将输入路径（Path会被转换为字符串形式）与原有值以“,”分隔进行拼接（并不会覆盖原有值），并保存至Job Configuration的属性INPUT_DIR（mapreduce.input.fileinputformat.inputdir）中。

 

 

而addInputPaths实际是对addInputPath的循环调用。

 

setInputPaths()实际是两个重载方法，用于“设置”一个（批）输入路径，该方法用于一次性调用，每一次调用都会覆盖之前的结果，源码如下：

 

 

该方法的最后会替换Job Configuration属性INPUT_DIR（mapreduce.input.fileinputformat.inputdir）的原有值。

 

 

 

这里所说的输入路径可以代表一个文件，也可以代表一个目录（该目录下的所有文件将全部作为输入数据），而且可以在输入路径中使用通配符或者使用“,”进行多个输入路径的拼接。

 

注意：目录中的内容（子目录）不会被递归处理。实际上目录中应仅包含文件，如果目录中包含子目录，这些子目录会被当作文件处理，从而引发异常。如果我们不需要递归目录，我们可以通过File Pattern或者Filter（见后）告知FileInputFormat仅仅选取指定目录中的文件；如果我们确实需要递归处理目录，则可以通过设置mapreduce.input.fileinputformat.input.dir.recursive为true实现。

 

有些时候我们还需要“过滤”输入路径中的一些文件，这可以通过方法setInputPathFilter()为FileInputFormat设置相应的过滤器实现，源码如下：

 

 

实际就是指定一个PathFilter（PathFilter的相关内容不再讨论范围）的具体实现类名称，保存于Job Configuration属性PATHFILTER_CLASS（mapreduce.input.pathFilter.class）中。

 

如果我们没有显示设置PathFilter，FileInputFormat会有一个默认的过滤器，用于过滤目录中的隐藏文件；如果我们显示设置PathFilter，则FileInputFormat的过滤器实则是一个过滤器链，而默认的过滤器会居于过滤器链的首部，优先被执行。

 

综上所述，FileInputFormat的输入路径和过滤器实际可以直接通过相应的属性值进行设置，如下图所求：

 

 

 

FileInputFormat生成切片的过程是由getSplits()方法实现的，核心逻辑及相关源码如下：

 

1. 确定切片大小的最小值与最大值；

 

 

最小值：getFormatMinSplitSize()与getMinSplitSize()两者之间的较大值。getFormatMinSplitSize()是FileInputFormat中的一个实例方法，默认返回值为1，即1字节；getMinSplitSize()返回值由属性mapreduce.input.fileinputformat.split.minsize决定，默认值为1，即1字节。如果没有特殊需要，最小值即为1字节。有些数据格式的文件对切片的最小大小是有要求的，如SequenceFile（具体可参考SequenceFile相关文档），这时就需要在FileInputFormat子类中重写getFormatMinSplitSize()方法来满足特定需求。

 

最大值：getMaxSplitSize()返回值由属性mapreduce.input.fileinputformat.split.maxsize决定，默认值为Long.MAX_VALUE。

 

2. 获取输入路径中的所有文件信息；

 

 

3. 迭代处理输入路径中的每一个文件，为每一个文件生成切片；

 

 

 

对于每一个文件而言，生成切片的过程大致可以概括为以下5个关键步骤：

 

获取文件的路径及长度（1）；

如果文件长度为0，则生成一个“空”的切片（5）；如果文件长度不为0，则获取文件的数据块信息（2）；

如果文件格式不可切片，则将整个文件生成一个切片（4）；如果文件格式可切片，则为该文件生成切片（3）。

 

其中文件格式是否支持切片，由FileInputFormat isSplitable()方法决定，默认返回值为true，即默认可切片。可以根据实际应用场景的不同，在FileInputFormat的子类中重写该方法，使返回值为false，达到禁止切片的功效，这样每一个Map Task会处理一个文件的全部数据。

 

在详细介绍第3步之前，需要先引入一个新的类FileSplit，它表示一个文件切片，包含的变量如下：

 

 

file：切片所引用的文件路径（名称）；

start：切片在文件中的起始偏移量；

length：切片大小;

hosts：由切片在文件中的起始偏移量、切片大小、文件数据块信息可以计算出切片所引用的数据块有哪些（切片大小可能大于HDFS的数据块大小），hosts中保存着这些数据块中的第一个数据块的副本位置（主机名），默认为3个主机名，MapReduce根据此值完成Map Task的调度；

hostInfos：相对于hosts中保存着的主机名，还保存着副本是否位于主机内存的信息。

 

 

下面介绍一个（可切片）文件切片的形成过程，大体也可以分为5个步骤：

 

step1

 

首先获取文件的数据块大小blockSize（这里也可以看出不同的文件，数据块大小也是可以不同的）；

 

然后根据数据块大小（blockSize）、切片最小值（minSize）、切片最大值（maxSize）计算文件对应的切片大小（splitSize），计算公式如下：

 

splitSize = max(minimumSize, min(maximumSize, blockSize))

 

step2

 

判断文件的剩余大小（未切片的大小）是否满足继续进行切片的条件：((double) bytesRemaining)/splitSize > SPLIT_SLOP为true，其中bytesRemaining初始值为文件长度length，SPLIT_SLOP值为1.1，且不可修改，即文件剩余大小需为切片大小（splitSize）的1.1倍才会继续切片。

 

step3

 

获取切片对应的数据块。一个切片根据切片大小的不同，可能会包含若干个数据块，这里将第一个数据块的副本位置作为切片的存储位置。

 

切片在文件中的起始偏移量的计算公式：

 

offset = （n - 1）* splitSize，n表示第几个切片

 

对于给定的切片的offset，getBlockIndex实际就是计算文件的哪个数据块的起止范围恰好包含offset，返回这个数据块在数据块列表（blkLocations）的下标，计算流程如下：

 

 

step4

 

根据下标对应的数据块信息构建一个FileSplit。根据FileSplit的信息，可以看出FileSplit并不实现保存数据，仅仅是通过文件名称、起始偏移量、大小关联数据，并将对应数据块的副本位置作为切片的存储位置进行Map Task的调度。

 

循环执行step2、step3、step4直到文件剩余大小无法满足切片条件。

 

step5

 

将文件的剩余部分构建一个FileSplit。