5分钟 Hadoop Shuffle 优化

上篇5分钟深入 Hadoop 的文章中，我们介绍了如何优化输入处理，让 Hadoop 达到更高的性能；另一个有可能让 Hadoop 性能实现质的飞越的过程是 Shuffle 阶段：Shuffle 阶段负责把 map output 传递到 reduce 阶段，深入理解这个阶段有助于我们回答面试中很多进阶问题，从而 really impress the interviewer. 本文就来详细讨论 Hadoop 中的 Shuffle 优化。

Shuffle 是什么？

讨论如何优化 Shuffle 之前，我们先来介绍 Shuffle 阶段具体发生了什么，从而对优化有个深入的理解。

Shuffle 阶段需要做哪些事情，完成哪些需求?

大家还记得 <5分钟搞懂 Hadoop> 那篇文章中举的例子吗？

我们从这个例子的图中可以看出，每个 map function 会输出一组 key value pair, Shuffle 阶段需要从所有 map host 上把相同的 key 的 key value pair 组合在一起，组合后传给 reduce host, 作为输入进入 reduce function 里。

那 Shuffle 阶段是如何完成这些功能的呢？

首先介绍新 Component：partitioner

所有 map function 产生的 key 可能有成百上千，经过 Shuffle 组合 key 的工作之后，依然是相同数目，而负责 reduce 工作的 host 可能只有几十个，几百个，那 Hadoop 的分配 key value pair 的策略是什么？

Partitioner component 负责计算哪些 key 应当被放到同一个 reduce 里。

举个栗子

使用默认的 partitioner: HashPartitioner，它会把 key 放进一个 hash function 里，然后得到结果。如果两个 key 的 hashed result 一样，他们的 key value pairs 就被放到同一个 reduce function 里。我们也把分配到同一个 reduce function 里的 key value pairs 叫做一个 reduce partition.

我们看到 hash function 最终产生多少不同的 result, 这个 Hadoop job 就会有多少个 reduce partition／reduce function，这些 reduce function 最终被JobTracker 分配到负责 reduce 的 host 上，进行处理。

在 Map 端的运作

Map function 的运行方式就是从 RecordReader 那边读出一个 input key value pair, 处理，然后把处理结果（通常也是 key value pair 形式）写到一个Hadoop maintained memory buffer 里，然后读取下一个 input key value pair.

Hadoop maintained memory buffer 里的 key value pair 按 key 值排序，并且按照 reduce partition 分到不同 partition 里（这就是 partitioner 被调用的时候）。一旦 memory buffer 满了，就会被 Hadoop 写到 file 里，这个过程叫 spill, 写出的 file 叫 spill file.

注意，这些 spill file 存在 map 所在 host 的 local disk 上，而不是我们之前介绍过的 HDFS.

随着 Map 不断运行，有可能有多个 spill file 被制造出来。当 Map 结束时，这些 spill file 会被 merge 起来——不是 merge 成一个 file，而是按 reduce partition 分成多个。

在 Reduce 端的运作：

由于 Map tasks 有可能在不同时间结束，所以 reduce tasks 没必要等所有 map tasks 都结束才开始。事实上，每个 reduce task 有一些 threads 专门负责从 map host copy map output（默认是5个，可以通过 $mapred.reduce.parallel.copies 参数设置）；考虑到网络的延迟问题，并行处理可以在一定程度上提高效率。

通过前面的学习，我们知道 Hadoop JobTracker 会根据输入数据位置安排 map tasks，但 reduce tasks 是不知道这种安排的。那么当 reduce task 需要从map host copy map output 时，它怎么知道 map host 的位置呢（URL／IP）?

其实当 Map tasks 成功结束时，他们会通知负责的 tasktracker, 然后消息通过 jobtracker 的 heartbeat 传给 jobtracker. 这样，对于每一个 job, jobtracker 知道 map output 和 map tasks 的关联。Reducer 内部有一个 thread 负责定期向 jobtracker 询问 map output 的位置，直到 reducer 得到所有它需要处理的 map output 的位置。

Reducer 的另一个 thread 会把拷贝过来的 map output file merge 成更大的 file. 如果 map task 被 configure 成需要对 map output 进行压缩，那 reduce 还要对 map 结果进行解压缩。当一个 reduce task 所有的 map output 都被拷贝到一个它的 host上时，reduce 就要开始对他们排序了。

排序并不是一次把所有 file 都排序，而是分几轮。每轮过后产生一个结果，然后再对结果排序。最后一轮就不用产生排序结果了，而是直接向 reduce 提供输入。这时，用户提供的 reduce function 就可以被调用了。输入就是 map task 产生的 key value pairs.

Hadoop Benefit

我们需要注意的是，Shuffle 阶段的功能是完完全全由 Hadoop framework 提供的，这里边没有任何用户的代码（即使我们有可能需要根据具体 Hadoop job 的特点配置一下这个阶段，但也非常方便）。

Shuffle 阶段应该说是整个 MapReduce Job 里需要处理问题最复杂，需要提高 performance 最多的地方。想象一下，公司里的 cluster 中可能有上百个 reducer 从上千个 mapper 那边高效率的拷贝处理数据，全靠 Hadoop 处理这个阶段，让我们可以轻轻松松就写写 map function 和 reduce function 就得到理想的效率，这是 Hadoop 广受欢迎的重要原因之一。

优化

在我们了解了 Hadoop 如何处理 shuffle 之后，我们可以通过配置一些 Hadoop Job 的参数调整 Hadoop shuffle performance：

• io.sort.mb 这个参数控制 map 端 memory buffer 大小，越大每次 map 要 spill 的 file 就越大，总 spill file 个数越少; 所以如果你的 Hadoop Job map output 很多，适当增加这个参数有助于 performance；
• io.sort.spill.percent 这个参数控制 map memory buffer 到达多少比例时开始 spill. 我们知道 spill 是写 IO 操作，所以需要时间。如果 percentage 太高，有可能当 spill 还没有完成时，map output 已经把 memory buffer 填满，这样影响 performance；同样，太低会造成太多 spill fie;
• tasktracker.http.threads 控制 map 端有多少个 thread 负责向 reduce 传送 map output. 本着并行计算的原则，可以适当调高;
• mapred.reduce.parallel.copies 这个参数控制 reduce 端有多少个 thread 去 copy map output. 本着并行计算的原则，可以适当调高;
• mapred.job.reduce.input.buffer.percent 控制 reduce host 上 JVM 中用于 merge map output 的比例。可以适当调高;
• io.sort.factor控制 reduce 端同时被 sort 的文件的个数。我们说 reduce 端 file sor t分批进行，这个参数就是每批有多少个。如果内存大，可以适当增加，以减少 sort 批次。

预告

• 通过之前两篇文章，大家应该对如何调整 Hadoop Job 的 performance 有了一个更深的了解。下一篇，我们将继续讨论面试中不可回避的 HDFS，也是Hadoop 系统的基石。欢迎大家关注。