探寻从 HDFS 到 Spark 的高效数据通道：以小文件输入为案例

为了保证高效的数据移动，locality 是大数据栈以及分布式应用程序所必须保证的性质，这一点在 Spark 中尤为明显。如果数据集大到不能保证完全放入内存，那就不能贸然使用 cache() 将数据固化到内存中。如果读取数据不能保证较好的 locality 性质的话，不论是对即席查询还是迭代计算都将面临输入瓶颈。而作为常用的分布式文件系统，HDFS 承担着数据存储、一致性保证等关键问题。HDFS 自开发之初就与 Google GFS 一脉相承，因此也继承了其无法较好的处理小文件的问题，但大量小文件输入又是分布式计算中常见场景。本文以小文件输入为案例，看看从 HDFS 到 Spark 的数据通道中到底发生了什么，并讨论如何设计有效的小文件输入。了解了这些话题，可以更高效的使用 Spark。

背景

MLlib 进展如火如荼，近期最令人振奋的消息莫过于 MLlib 对 sparse vector 的支持，以及随之而来的一系列重构和改进工作。机器学习一般算法的输入是训练集和测试集，通常来说是 (label, key : value) 这样的序对。对于这种输入，直接使用 SparkContext 提供的 textFile() 接口就好了，MLlib 内部会转换成 LabeledPoint 类。但 MLlib 还缺少图模型算法，如 LDA。LDA (Latent Allocation Dirichlet) 算法常用来获取文档集合的主题，是机器学习中广泛应用的算法，其输入格式和核心组件与常见的机器学习分类、聚类算法不同。

两个月之前，笔者有一份差不多要完成的基于 Spark 的并行 LDA 算法准备提交给 Spark 社区，同时也在准备酝酿已久的学术论文。当笔者完成了 LDA 算法的核心模块 Gibbs sampling 之后，突然发觉要想实现一个“可用的“LDA 算法，不仅仅是一个核心功能这么简单，还牵扯到许多零碎的事情。所谓零碎的事情，其实并不简单。机器学习算法就是这样，学起来难，但是真正懂了之后发现核心算法特别简单，预处理又非常难。总之，机器学习算法学起来难的地方做起来简单，但是学起来简单的地方, 并不见得很快就能做好。

大部分的零碎工作在语料库的预处理和后续输出的模型的使用上，这些零碎的工作机器学习者们都不怎么注重，因为书本上很少会讲到这些知识。就拿后者来说，模型后续使用这件事儿看起来小，其实不然，这关系到机器学习算法的实际运用能力。我们做模型的最终目的除了发论文之外还是想让它对现实生活产生影响。学术派关注模型多，但对学术和工业的结合看的相对少一点，线下模型如何轻松部署？模型可否增量训练？模型的训练和使用是否可以同步进行？是否可以做到对模型的在线查询？这都是将机器学习“搬出实验室”的关键问题。这类问题在 Strata 大会上有很多工业界人士做了很好的讲解，比如这里。

闲言莫谈，回到语料库的预处理工作。关于分词的问题不多谈，笔者学习ScalaNLP 的做法，直接在Lucene 的分词实现上裹了一层Scala 的接口。但是在语料集的输入上花了很多时间。Spark 目前所有的标准输入接口是SparkContext 类中提供的textFile(path, miniSplits) 接口，但该接口不适合语料库的直接输入，因为这是一个文本行处理函数，每次只能操作其中的一行文本。而LDA 更期待的输入格式是Key-Value 对，其中key 是文件的绝对路径（便于分辨和去重），value 是文件的全部内容。由于Spark 下层多使用HDFS 作为输入，因此笔者打算自己定制InputFormat。

LDA 应用场景

首先得说明一些问题。LDA 的实际使用场景有二：

第一种是在实验室环境下使用，这是最直观的情况。例如你有一堆小文件存在本地磁盘上，即你的语料库。可能你想直接把它们上传到 HDFS，或者在每台机器的磁盘上仍一份，甚至直接放在当前机器的本地磁盘好了（这种情况下不是真正的分布式，所有的 Spark executor 只会在你当前机器上启动），之后打开 Spark 调用其中的 LDA 算法。如果你只是打算做个实验，这样就足够了。换言之，这是一种 offline 的训练方式。

第二种情况是工业应用，你可能不会有一堆离线的语料库，而是有一个流式管道，语料文本源源不断地传递过来，如 twitter/weibo feed 等。或许你可以把这些数据放到 HDFS 或 HBase 上，也有可能直接处理流数据，而不管最终存储。

这是完全不同的应用场景，针对不同的场景要有不同的处理方式。不论是实验室环境下的尝试，还是工业应用，两者都很重要。本文只涉及 offline 的数据处理方法，因为 offline 的数据处理才更加需求让数据经过 HDFS。

离线 LDA 训练

离线场景下或许我们不必理会语料集预处理的过程，直接交给最终用户好了。用户将语料集转换成你指定的样式，之后将预处理结果上载到 HDFS，这样你的 LDA 程序可以直接访问这部分数据，而我们要做的只是规定好输入的样式，妙不可言。我们舒服了，用户吃些苦头。例如我们指定用户输入文件的每一行是一个完整的文件内容，开头处以 Tab 分割作为文件名。这样我们可以直接调用 textFile() 接口，自己切分一下就可以得到”文件名–文件内容”这个 KV 对。值得一提的是，这种离线场景下看似不好用的方法，在工业界线上训练过程中反倒可能有好的效果。比如一次记录过来就是一个 KV 对，这样就省去了这一步输入的处理。

或许我们可以进一步帮帮用户。咱们写一个预处理程序，不论是串行的还是并行的，帮助用户进行预处理，Mahout 就是这么做的。这种情况下，可能需要最终用户写一个 ad-hoc 的 shell 脚本组织所有的工作流和数据流。Mahout 中的 dirTosequentialFile 就是把本地磁盘或者 HDFS 上的目录读入，将其中的小文件合并在一起转换成一个 sequential file。

但是，笔者觉得最好的方法还是将预处理过程与 LDA 训练过程融合起来，不要让用户做这么多工作就能调用 Spark 上的 LDA，用户只需要指定文件路径即可。这时我们必须提供函数将语料库所有的文本和文件名读入。CombineFileInputFormat 比较适合处理小文件，因此最初笔者实现了一个 CombineFileInputFormat，一个 CombineFileRecordReader，一个 FileLineWritable 以及一个类似于 textFile() 的接口。

Interface exposed to end-user - SparkContext.scala
 def wholeTextFiles(path: String): RDD[(String, String)] = {
   newAPIHadoopFile(
     path,
     classOf[WholeTextFileInputFormat],
     classOf[String],
     classOf[String])
 }

要注意的是，虽然笔者这么做了，但这并不代表小文件输入的最佳实践。实际上，最佳实践是不要使用小文件。因为将大量的小文件放到 HDFS 上是比较糟糕的，不仅将 block 用满率降低，还会占满 NameNode 上面的索引。这里只是讨论一种可行的方案。

分析几个问题。首先是 HDFS 中 block 的大小。我们在这里称之“小文件”的文件究竟有多“小”？是否会超过 HDFS 的 block 大小？答案是肯定的。在这种情况下，如果我们按照 block 位单位读取数据，那么我们就要自己处理同一个文件的 block 拼接的问题，尤其是文件由多字节字符组成的时候，如 UTF 编码的字符，很可能在一个字符中间被切断。如果不能正确的拼接各个 block，会出现乱码的情况。

重点是，考虑到性能问题，我们不希望有额外的网络传输开销存在，尤其是不必要的网络传输。我们希望同一个文件的 block 都在同一个节点上，这样在合并这些 block 的时候就完全不会出现机器之间数据的网络传输。HDFS 里面很讨厌的一点是，这里有两套极为相同的 API，分别在 mapred 和 mapreduce 两个包下，稍不注意用错了 API 就会有一种很抓狂的感觉。笔者最初为了兼容 Spark 中 HadoopRDD 的接口，用了 mapred 的这套 API，其中的 CombineFileRecordReader 中的 isSplitable() 函数是不起作用的，因为如果不修改 CombineFileRecordReader 本身的代码就无法阻止一个文件的多个 block 分配到不同的 split 中的情况。一旦这件事情发生，那拼接一个文件的时候就无法阻止 shuffle 的发生。

在线 LDA 训练

现在来看看在线训练。注意线上产品不应该使用上述方式运行。当然，如果不顾及线上模型训练，认为模型可以线下训练好的话另当别论。数据处理部门的人是不会把大量原始文本存储到本地磁盘，之后再将数据上传到服务器处理的。在我看来，大数据就应该放到合适的地方去。这种场景下，原始文本或者网页应该存储到一种 KV 存储中，例如 HBase（Facebook 在其论文 Analysis of HDFS Under HBase: A Facebook Messages Case Study中详述了 HDFS 之上的 HBase 性能问题，值得一看。）。除此之外，HDF5 也是种不错的选择。

网络传输来自哪里

笔者一直在讨论避免网络传输开销的问题，那么网络传输到底出现在哪里从而导致的不可避免呢？

首先，文件大小超过单个 block 大小的文件不免被切分，不论是 ASCII 编码的文件，还是 UTF 这种多字节编码的文件，都需要一个 join 的过程。最好的情况下，我们期望同一个文件所有的 block 都从同一个机器上读取，这样可以避免网络传输。

第二，出于效率的考量，mapred 包中的 CombineFileInputFormat 不能保证这一点。这是因为每个 block 都会有副本的存在。为了保证数据读取的高效，同一个文件的不同 block 可能读取不同机器上的副本。同时由于单个 split 大小的限制，同一个机器上的 block 也可能分到不同的 split 里面。正是由于 HDFS 多副本容错的特性，导致同个文件不同的 block 甚至可能在任何一个位置被读取。

自定义 Partitioner 怎么样

在笔者看来，Spark 之所以好用的原因之一就是可以简单地定制 partition 方法。使用自定制的 partitioner 来重新安排我们 KV 对的存储。然而，定制化的 partitioner 最大的作用是在迭代地进行 RDD join 的时候，正如 Spark PageRank 所展示的那样。如果是一次性的使用，有点得不偿失，因为第一次的 shuffle 在所难免。

Hadoop locality 全揭秘

为了更好的了解 Hadoop I/O 保持良好 locality 的秘密，我们要深入看一看 mapred 包中的 InputFormat 实现。我们选择 FileInputFormat 作为突破口，因为这也是 spark 的“重点使用对象”。首先要记住这些后面会经常用到的概念：rack、node、file、block、replica。数据中心通常由一堆堆 rack 组成，rack 是同一个机架中的机器。由于同个 rack 之间网络状况通常都是非常好的，因此考虑本地性的时候通常也将同个 rack 中的数据算作本地数据。一个 rack 由多个 node 组成，这里的 node 特指作为 DataNode 的机器。HDFS 上每个文件包含多个 block，每个 block 有一些副本（通常是三个）。要注意的是 Hadoop 的 worker 可能包含所有的 DataNode 节点，当然也会出现不匹配的现象，即有些机器仅仅是 DataNode 节点而非 Hadoop worker，反之亦然。同时也要注意，考虑到鲁棒性，每个 block 的三副本通常都是当前 node 一个，本 rack 其他机器一个，其他 rack 上一个。

把 Hadoop worker 也考虑进来后，问题稍显复杂。程序可能分布在多个 worker 上，数据分布在多个 DataNode 上。因此，问题抽象成如何在 worker 和 DataNode 之间做多对一的映射（一个 worker 会可能处理多个 DataNode 上的数据，而一份数据通常只要一个 worker 处理就好了），使得读取 HDFS 文件时造成的网络开销最小？换句话说，读取文件整体耗时最少。

这件事儿不是很容易，因为应用程序和数据之间还隔着好多层。程序所直接接触的就是文件名。一个文件被分为多个 block，每个 block 可能存在于每一个数据节点上，其副本存在于其他节点上，不同的节点还属于不同的 rack。我们先从程序开始看起。以 Spark 为例，我们调用 hadoopRDD = sc.textFile(path) 告诉 Spark 开始读取 path 中的数据。这个 path 可能是一个本地文件路径，更常见的是 HDFS 路径。为了分布式处理的要求，hadoopRDD 通常情况下是被切分的。那么，其 partition 的信息来自何处呢？答案就是 HDFS 中的 split，更确切的说是 FileSplit，其在 FileInputFormat 中被用到。FileSplit 就是这样一种程序和 block 之间的映射。

每个 FileSplit 都是一个 block 集合，里面的 block 会在同一个 worker 上的同一个程序读出，因此也理所当然作为一个 partition。为了保持同一个 split 中 block 的本地性，FileSplit 花了大力气把合适的 block 放到一起。例如贡献度计算，以及 node-block、rack-block 之间的双向链表等。现在我们把之前的程序 -block 映射问题退化成 split-block 的映射问题。

Node/Rack 贡献度计算

假设我们有一个 split，其中有三个 block，这三个 block 来自 8 个节点。8 个节点属于 4 个不同的 rack，每个 block 有三个副本。假设这三个 block 的大小分别为 100、150、75。这种情况下怎么安排“最佳地点”？即该 split 应该在哪个 worker 上计算？

首先，我们一致同意的一点是“最佳地点”应该是所有 node 的子集。在我们的例子中，这个集合是 [h1 — h8] 这八个节点。怎样对这个集合进行排序，依次找到“最佳地点”、“次佳地点”？

对节点集合进行排序有两种方法，分别是考虑 rack 的信息和不考虑 rack 的信息。上文说过，可以将同一个 rack 中的 block 也算做本地的 block。要想排序，先要确定准则，即什么才是“好”。参照上图，我们定义一个“effective size”的概念，这是说在本节点上，存在多少比特有效的数据可供读取。同样的，rack 的 effective size 就是该 rack 上所有的有效读取的比特。注意，并非是将该节点有的 block 以及字节数加起来这么简单，这里的“有效”是指的有区分度的字节数。例如，Rack4 有两个 block，每个 block 的大小都是 75，但 Rack4 的 effective size 只有 75，并非 150，因为这两个 block 具有相同的内容，他们互为副本。

考虑到 rack 的计算方式就是将 rack 看作跟 node 同等的位置，计算 effective size 之后，可得如下顺序：

1. Rack 2 (250)

1. h4 (150)
2. h3 (100)

2. Rack 1 (175)

1. h1 (175)
2. h2 (100)

3. Rack 3 (150)

1. h5 (150)
2. h6 (150)

4. Rack 4 (75)

1. h7 (75)
2. h8 (75)

因此，优先顺序是h4 > h3 > h1 > h2 > h5 > h6 > h7 > h8。

不考虑 rack 的方法更简单，节点排序的结果为：

1. h1 (175)
2. h4 (150)
3. h5 (150)
4. h6 (150)
5. h2 (100)
6. h3 (100)
7. h7 (75)
8. h8 (75)

其优先顺序为 h1 > h4 > h5 > h6 > h2 > h3 > h7 > h8。

更多细节参见 Hadoop 的测试用例： https://github.com/apache/hadoop-common/blob/release-1.0.4/src/test/org/apache/hadoop/mapred/TestGetSplitHosts.java

双向链表

CombineFileInputFormat 选择了另外的方式保持 locality 的性质，它使用双向链表将 block 串在一起，之后先是逐节点扫描 block，再次是逐 rack 扫描 node，最后剩下来的“残片”丢到最后一堆处理，这样最大限度的保证 locality，并且维持 partition 的大小尽可能平衡。缺点就是出现跨 block 的文件的情况下，同一个文件的 block 很有可能落到不同的 partition 中。这里的陷阱是，在 Hadoop 老 API 中，isSplitable() 函数不能起到保持同一个文件内容在同一个 partition 中的作用，而在新 API 中反倒有这个功能了。各位使用的时候可要睁大眼睛。顺便说一句，新 API 虽然加入了这个功能，但是不切分文件的情况下，在保持 locality 和 partition 均衡的性质上可就没老 API 好了。无论如何，这些 trade-off 总是逃不掉的。

 
Double linked lists sweep for constructing split - CombineFileInputFormat.java
// mapping from a rack name to the list of blocks it has
HashMap<String, List<OneBlockInfo>> rackToBlocks =
new HashMap<String, List<OneBlockInfo>>();
// mapping from a block to the nodes on which it has replicas
HashMap<OneBlockInfo, String[]> blockToNodes =
new HashMap<OneBlockInfo, String[]>();
// mapping from a node to the list of blocks that it contains
HashMap<String, List<OneBlockInfo>> nodeToBlocks =
new HashMap<String, List<OneBlockInfo>>();
...
// process all nodes and create splits that are local
// to a node. 
for (Iterator<Map.Entry<String,
List<OneBlockInfo>>> iter = nodeToBlocks.entrySet().iterator();
iter.hasNext();) {

如何读取

聊了这么多，我们总算清楚 MapReduce 类的程序如何在组织 split 的时候保持良好的 block 的本地性了。我们很开心的获得其中的“最佳地点”，并将这个信息传递给 spark 的 partition。下一步工作就是 Spark 根据“最佳地点”，如上例中的节点 h4，启动 worker 上的处理进程 / 线程开始读取数据了。现在 h4 启动了 spark 的 executor 开始处理 split 中的 block。但是稍等，h4 怎么知道从哪个节点上获得每个 block 呢？要知道，每个 block 有三个副本呢！具体读取该 block 的哪个副本这个信息并未传递给 partition。

从 RecordReader 开始，我们再来一步步还原数据读取的过程。有了上面的基础，这次的旅程应该很快了。以笔者写的 BatchFileRecordReader 为例：

 
Constructer of BatchFileRecoderReader - BatchFileRecorderReader.java
   public BatchFileRecordReader(
           CombineFileSplit split,
           Configuration conf,
           Reporter reporter,
           Integer index)
           throws IOException {
       path = split.getPath(index);
       startOffset = split.getOffset(index);
       pos = startOffset;
       end = startOffset + split.getLength(index);
 
       FileSystem fs = path.getFileSystem(conf);
       fileIn = fs.open(path);
       fileIn.seek(startOffset);
 
       totalMemory = Runtime.getRuntime().totalMemory();
   }

在上面的代码中，我们从 split 中拿到了 path，注意这里的 path 是当前文件路径，可不是 block 路径。有了它，我们可以拿到一个 fileIn，其类型为 FSDataInputStream。之后我们 seek 到这个 block 开始的位置，称作 startOffset。等一下，我们根本没用到“最佳地点”的信息，是不是很奇怪呢？我们之前花了大量力气拿到的信息，这里没有用到。

这里需要记住的是，目前为止我们获得的 split 信息只是为每个 block 分配了计算节点，仅此而已。如何读取由别的代码控制。那么再来看看 FSDataInputStream，这里也没有太多东西，只有一些看上去没啥用的代码。

 
FSDataInputStream.java
public class FSDataInputStream extends DataInputStream
   implements Seekable, PositionedReadable, Closeable {
 
   public FSDataInputStream(InputStream in)
       throws IOException {
       super(in);
       if( !(in instanceof Seekable) || !(in instanceof PositionedReadable) ) {
           throw new IllegalArgumentException(
           "In is not an instance of Seekable or PositionedReadable");
       }
   }
 
   public synchronized void seek(long desired) throws IOException {
       ((Seekable)in).seek(desired);
   }
}

好吧，我们另觅他途。注意之前 fileIn 是由 fs.open() 这个调用获得的，在 HDFS 的场景下，这个 fs 其实是 DistributedFileSystem，即常说的 DFS。结果我们在 DFS 中找到了由 DFSInputStream 包装成的 FSDataInputStream，前者在 DFSClient 中实现。我们所期待的函数是 blockSeekTo()，这个函数负责给定偏移量之后找到合适的 block。之后它会找到最优的 DataNode，并从中读取数据。

 
Find an appropriate block and select a DataNode - DFSClient.java
   DatanodeInfo chosenNode = null;
   int refetchToken = 1; // only need to get a new access token once
   while (true) {
       //
       // Compute desired block
       //
       LocatedBlock targetBlock = getBlockAt(target, true);
       assert (target==this.pos) : "Wrong postion " + pos + " expect " + target;
       long offsetIntoBlock = target - targetBlock.getStartOffset();
 
       DNAddrPair retval = chooseDataNode(targetBlock);
       chosenNode = retval.info;
       InetSocketAddress targetAddr = retval.addr;
   }

这其中最重要的函数是 chooseDataNode()，它非常简单，只是从一个 DataNode 列表中选择第一个 node。如果第一个 node 连接不上，再找第二个，依次类推。bestNode() 函数中的注释说这里的 node 列表已经按照优先规则排序好的了。很奇怪，这是在什么时候排序的呢？

实际上，在这个文件首次打开的时候就已经排序好了。参见 openInfo() 函数，它调用 callGetBlockLocations() 函数进行排序。后者在 NameNode 中的 getBlockLocations() 中查询信息。可以看到它调用了 clusterMap 中的 pseudoSortByDistance() 进行排序。至此，我们获得了 Hadoop 为应用保持数据本地性的全景。

 
Get block locations and sorted in the priority order - FSNamesystem.java
   LocatedBlocks getBlockLocations(String clientMachine, String src,
       long offset, long length) throws IOException {
       LocatedBlocks blocks = getBlockLocations(src, offset, length, true, true);
       if (blocks != null) {
           //sort the blocks
           DatanodeDescriptor client = host2DataNodeMap.getDatanodeByHost(
               clientMachine);
           for (LocatedBlock b : blocks.getLocatedBlocks()) {
               clusterMap.pseudoSortByDistance(client, b.getLocations());
           }
       }
       return blocks;
   }

结语

分布式数据并行环境下，保持数据的本地性是非常重要的内容，事关分布式系统性能高下。要想更好的了解 Spark 是怎么运作的，输入也许是很重要的一个环节。举一个小例子，你或许有心情在一台不错的机器上使用 Spark 处理 100GB 的数据。按理说这不应算作多大的应用场景，但如果不仔细调整一下你的输入的话，你会发现 Spark 甚至会在这台机器上切分上千个 partition 来并行处理这份数据。而这上千个 partition 随便来一个 shuffle 造成的百万量级的 shuffle 数据交换会把 Spark 性能拖死。实际上，调用 Hadoop 的 API 访问本地磁盘的默认块大小为 32MB，据其分块策略，当然会产生上千个 partition。另外，如果你本地是一堆小文件，如 LDA 的语料库，你会发现 Spark 甚至会为每个文件分配一个或多个 partition！所以，这下你应该知道为什么有时简单的 Spark 程序也会非常慢了吧。

本文为了解决 LDA 小文件输入的问题，一步步揭开 HDFS 与 Spark 的数据通道的故事。总结来看，为了分布式使用各个机器，HDFS 读取的时候将数据分成了各个分块，为了防止 straggler 的产生，MapReduce 的读取模块会尽量保证各个分块在每台机器上的大小和个数均衡。为了保证较好的 locality，Spark 获取 preferredLocation 信息，尽量保证在临近的机器上读取所需的数据。为了合理读取小文件，CombineFileInputFormat 合理安排小文件分片，既要保证数据在各个分块中均衡，又不能切断单个文件。为了保证 HDFS 与 Spark 之间的高效数据通道，正可谓”无所不用其极”。

作者简介

尹绪森，Intel 实习生，熟悉并热爱机器学习相关内容，对自然语言处理、推荐系统等有所涉猎。目前致力于机器学习算法并行、凸优化层面的算法优化问题，以及大数据平台性能调优。对 Spark、Mahout、GraphLab 等开源项目有所尝试和理解，并希望从优化层向下，系统层向上对并行算法及平台做出贡献。

感谢辛湜对本文的审校。

感谢包研对本文的策划。

给InfoQ 中文站投稿或者参与内容翻译工作，请邮件至 editors@cn.infoq.com 。也欢迎大家通过新浪微博（ @InfoQ ）或者腾讯微博（ @InfoQ ）关注我们，并与我们的编辑和其他读者朋友交流。