深刻浅出Hadoop数据流（1）

1. 首先是一些术语的说明。MapReduce做业(job)是客户端执行的单位：它包括输入数据、MapReduce程序和配置信息。Hadoop经过把做业分红若干个小任务(task)来工做，其包括两种类型的任务：map任务和reduce任务。

2. 有两种类型的节点控制着做业执行过程：jobtracker和多个tasktracker。jobtracker经过调度任务在tasktracker上运行，来协调全部运行在系统上的做业。Tasktracker运行任务的同时，把进度报告传送到jobtracker，jobtracker则记录着每项任务的总体进展状况。若是其中一个任务失败，jobtracker能够从新调度任务到另一个tasktracker。Hadoop把输入数据划分红等长的小数据发送到MapReduce，称为输入分片(input split)或分片。Hadoop为每一个分片(split)建立一个map任务，由它来运行用户自定义的map函数来分析每一个分片中的记录。

3. 拥有许多分片就意味着处理每一个分片的时间与处理整个输入的时间相比是比较小的。所以，若是咱们并行处理每一个分片，且分片是小块的数据，那么处理过程将有一个更好的负载平衡，由于更快的计算机将可以比一台速度较慢的机器在做业过程当中处理完比例更多的数据分片。即便是相同的机器，没有处理的或其余同时运行的做业也会使负载平衡得以实现，而且在分片变得更细时，负载平衡质量也会更佳。

4. 另外一方面，若是分片过小，那么管理分片的总时间和map任务建立的总时间将决定做业的执行的总时间。对于大多数做业，一个理想的分片大小每每是一个HDFS块的大小，默认是64 MB，虽然这能够根据集群进行调整(对于全部新建文件)或在新建每一个文件时具体进行指定。

5. map任务的执行节点和输入数据的存储节点是同一节点，Hadoop的性能达到最佳。这就是所谓的data locality optimization(数据局部性优化)。如今咱们应该清楚为何最佳分片的大小与块大小相同：它是最大的可保证存储在单个节点上的数据量。若是分区跨越两个块，那么对于任何一个HDFS节点而言，基本不可能同时存储这两数据块，所以此分布的某部分必须经过网络传输到节点，这与使用本地数据运行map任务相比，显然效率更低。

6. map任务把输出写入本地硬盘，而不是HDFS。这是为何？由于map的输出做为中间输出：而中间输出则被reduce任务处理后产生最终的输出，一旦做业完成，map的输出就能够删除了。所以，把它及其副本存储在HDFS中，不免有些小题大作。若是该节点上运行的map任务在map输出给reduce任务处理以前崩溃，那么Hadoop将在另外一个节点上从新运行map任务以再次建立map的输出。

7. reduce任务并不具有数据本地读取的优点-- 一个单一的reduce任务的输入每每来自于全部mapper的输出。在本例中，咱们有一个单独的reduce任务，其输入是由全部map任务的输出组成的。所以，有序map的输出必须经过网络传输到reduce任务运行的节点，并在那里进行合并，而后传递到用户定义的reduce函数中。为增长其可靠性，reduce的输出一般存储在HDFS中。如第3章所述，对于每一个reduce输出的HDFS块，第一个副本存储在本地节点上，其余副本存储在其余机架节点中。所以，编写reduce的输出确实十分占用网络带宽，可是只和正常的HDFS写管线的消耗同样。

8. 一个单一的reduce 任务的整个数据流如图2-2所示。虚线框表示节点，虚线箭头表示数据传输到一个节点上，而实线的箭头表示节点之间的数据传输。

9. reduce任务的数目并非由输入的大小来决定，而是单独具体指定的。在第7章的7.1节中，将介绍如何为一个给定的做业选择reduce任务数量。

10. 若是有多个reducer，map任务会对其输出进行分区，为每一个reduce任务建立一个分区(partition)。每一个分区包含许多键(及其关联的值)，但每一个键的记录都在同一个分区中。分区能够经过用户定义的partitioner来控制，但一般是用默认的分区工具，它使用的是hash函数来造成"木桶"键/值，这种方法效率很高。

11. 通常状况下，多个reduce任务的数据流如图2-3所示。此图清楚地代表了map和reduce任务之间的数据流为何要称为"shuffle"(洗牌)，由于每一个reduce任务的输入都由许多map任务来提供。shuffle其实比此图所显示的更复杂，而且调整它可能会对做业的执行时间产生很大的影响，详见6.4节。

（点击查看大图）图2-2：MapReduce中单一reduce任务的数据流图

1.  

（点击查看大图）图2-3：多个reduce任务的MapReduce数据流

2.4.1  数据流（2）

最后，也有可能不存在reduce任务，不须要shuffle的时候，这样的状况是可能的，由于处理能够并行进行(第7章有几个例子讨论了这个问题)。在这种状况下，惟一的非本地节点数据传输是当map任务写入到HDFS中(见图2-4)。

图2-4：MapReduce中没有reduce任务的数据流

集群的可用带宽限制了MapReduce做业的数量，所以map和reduce任务之间数据传输的代价是最小的。Hadoop容许用户声明一个combiner，运行在map的输出上-- 该函数的输出做为reduce函数的输入。因为combiner是一个优化方法，因此Hadoop不保证对于某个map的输出记录是否调用该方法，调用该方法多少次。换言之，不调用该方法或者调用该方法屡次，reducer的输出结果都同样。

combiner的规则限制着可用的函数类型。咱们将用一个例子来巧妙地加以说明。之前面的最高气温例子为例，1950年的读数由两个map处理(由于它们在不一样的分片中)。假设第一个map的输出以下：

1.  (1950, 0)  

2.  (1950, 20)  

3.  (1950, 10) 

第二个map的输出以下：

1.  (1950, 25)  

2.  (1950, 15) 

reduce函数再调用时被传入如下数字：

1.  (1950, [0, 20, 10, 25, 15]) 

由于25是输入值中的最大数，因此输出以下：

1.  (1950, 25) 

咱们能够用combiner，像reduce函数那样，为每一个map输出找到最高气温。reduce函数被调用时将被传入以下数值：

1.  (1950, [20, 25]) 

然而，reduce输出的结果和之前同样。更简单地说，咱们能够像下面这样，对本例中的气温值进行以下函数调用：

1.  max(0, 20, 10, 25, 15) = max(max(0, 20, 10), 
max(25, 15)) = max(20, 25) = 25 

并不是全部函数都有此属性。例如，若是咱们计算平均气温，便不能用mean做为combiner，由于：

1.  mean(0, 20, 10, 25, 15) = 14 

可是：

1.  mean(mean(0, 20, 10), mean(25, 15)) = mean(10, 20) = 15 

combiner并不能取代reduce函数。(为何呢？reduce函数仍然须要处理来自不一样的map给出的相同键记录。)但它能够帮助减小map和reduce之间的数据传输量，而正由于此，是否在MapReduce做业中使用combiner是须要慎重考虑的。