本文首发于知乎专栏：进击的云计算
本书翻译于：Designing Distributed Systems
望小伙伴们多多支持原版。

前面的章节描述了一系列将队列拆分和连接在一起以实现更复杂批处理的模式，复制和生成多个不同的输出是批处理的重要组成部分，但有时将多个输出合并到一起以生成某种聚合输出也同样很重要，如图1所示。

这种聚合最典型的例子是MapReduce模式中的Reduce部分，很容易看出，Map步骤是将作业队列进行分割，而Reduce步骤是进行协调处理，最后将大量的输出聚合为单个输出。但是，批处理有许多不同的聚合模式，本文除了讨论真实的应用程序之外，还讨论了其中的一些模式。

Join（or Barrier Synchronization）

---

在前几篇文章中，我们看到了将作业分开并将其分散在多个节点上的模式，特别是，我们看到了一个分片作业队列如何将作业分配给许多不同的作业队列分片。但是，有时在处理作业流时，有必要在进入下一个作业流阶段之前保证作业集合的完整性。

这样做的一种方法是在前一篇文章中提到的，即将多个队列合并在一起，然而，合并只是将两个作业队列的输出合并为一个作业队列来进行其他处理。虽然在某些情况下，Merge模式已经足够了，但是并不能确保开始处理之前的数据完整性，这意味着不能保证正在执行的处理的完整性，对于已处理的作业，也没有办法计算汇总统计信息。

我们需要一个更加强大的、可以进行批数据处理的协调模式，这种模式就是Join模式。如图2所示，Join类似于加入一个线程，其基本思想是所有的作业都是并行处理的，但是在所有并行处理的作业项都全部完成时，作业项才会从Join中释放出来，这也通常被称为并发编程中的Barrier Synchronization。

通过Join模式进行协调可以确保在执行某种聚合阶段之前没有数据丢失（例如，求一组数的总和），Join的值是确保集合中所有的数据都存在。Join模式的缺点是它要求所有的数据都要在之前的阶段都已经处理好，然后才能进行后续的计算，这会降低批处理作业流程中可能的并行性，从而增加整个作业流的整体延迟。

Reduce

---

如果对作业队列进行分片处理是典型的map／reduce算法的map阶段，那么剩下的就是reduce阶段。Reduce是协调批处理模式的一个例子，它将不同批操作产生的并行输出组合成的不同数据。

但是，与前面介绍的Join模式相比，Reduce的目标不是等所有数据都被处理完毕，而是乐观地将所有并行数据项合并到一起。

使用Reduce模式时，Reduce中的每个步骤都会将几个不同的输出合并为一个输出，因为这个阶段它减少了输出的总数，所以它被称为“Reduce”。此外，他还将数据从完整的数据项简化为所需的代表性数据。由于Reduce阶段是在一定范围的输入上进行运算并产生输出，因此该阶段可以根据需要重复多次，以便成功的将多个输出减少为单个输出。

这与Join模式形成了鲜明的对比。因为与Join不同的是，就算Map阶段还在进行处理，Reduce阶段依旧可以开始并行的处理。当然，为了保证数据的完整性，所有的数据最终都要被处理，但是能够早些开始处理就意味着批处理的执行速度更快。

Sum

一种不同的Reduce形式是计算不同值的总和，它不是简单地为每个值计算一个值，而是实际上将原始输出数据中存在的值相加。

例如你想统计美国的人口总数，假设你是通过计算每个城镇的人口数量，然后将它们加在一起的方式来实现的。

第一步是将统计全国人口这件事情分解成统计每个城镇的人口数量，但即便并行的进行处理，也需要很长时间才能统计每个城镇的人数，因此我们需要将统计城镇人口这个作业进行第二次划分，按照县来划分。

按照县来统计时，只要有统计结果出来，我们就可以进行Reduce的操作，在这种情况下，Reduce甚至不需要知道我们把作业分成了什么样，只需要简单的将两个或更多的输出项汇总在一起产生新的输出就足够了。这就像计算一样，Reduce阶段可以执行任意的次数，只要每个时间间隔运行相同的代码，最后就会有一个包含美国人口总数的输出。最重要的是，几乎所有的计算都是并行进行的。

Histogram

我们依旧思考通过map／shard和reduce的方式来统计美国人口总数的例子，同时我们也希望建立一个普通的美国家庭的模式。要做到这一点，我们要制定一个家庭规模的直方图，也就是估计有0-10个孩子的家庭总数的模式。我们将完全按照之前的方式来执行我们的作业分片（实际上，我们可能会使用相同的worker）。

但是，这一次数据收集阶段的输出是每个城镇的直方图。

0：15％
1：25％
2：50％
3：10％
4：5％

从前面的例子中，我们可以看到，如果我们使用Reduce模式，我们应该能够将所有的直方图聚合起来，形成完整的美国人口图。乍一看，似乎很难理解如何合并这些直方图，但是当与示例中的总体数据相结合时，我们可以看到，如果我们将每个直方图与其相对应的人口数目相乘，那么我们可以获取每个合并项的总人口数，然后再将这个新的总数除以合并后的总数，就是可以得到新的输出。鉴于此，我们可以根据需要多次运行Reduce阶段，直到产生单个输出为止。

我们通过标记和处理一组图像的例子来更深入的了解协调批处理是如何来完成更大的批处理任务。假设我们拥有大量的高峰时段的高速公路图像，我们想要计算道路上汽车、卡车和摩托车的数量，以及每辆车的颜色分布，同时也假设有一个步骤可以模糊掉所有汽车的车牌号。

图像以一堆HTTPS URL的形式发送给我们，其中每个URL指向一个原始的图像。流水线的第一个阶段就是寻找并且模糊掉车牌信息，为了对作业队列中的每个任务进行简化，我们将使用一个worker来负责检测车牌在图片中的位置，另外一个worker负责模糊该位置的车牌信息。我们将使用之前文章中描述的multi-worker模式来将这两个不同的worker容器合并到一个容器组中，这样的分离可以提高重用性，比如用来模糊图像的worker可以重用来模糊其他的输出（例如，模糊人脸）。

此外，为了确保可靠性并最大化并行处理，我们将通过多个worker队列来将这些图片进行分片，图3显示了完整的作业流程。

一旦每张照片模糊成功，我们就会将其上传到不同的位置，然后我们将原件删除，但是，为了防止某种在灾难性事故的发生，我们在所有图片都进行模糊操作之前并不想删除原始的图片。因此，为了等待所有的照片都模糊完成，我们使用Join模式将所有分片模糊作业队列的输出合并到一个队列当中，只有当所有的分片任务全部完成之后才会释放这些作业项。

现在我们准备删除原始图片，并开始对车辆的车型和颜色进行检测。但是，我们想将流水线的吞吐量最大化，所以我们将使用前一篇文章介绍的Copier模式将作业队列项复制到两个不同的队列中（如图4所示）：

• 删除原始图像的作业队列
• 识别车辆类型（汽车、卡车、摩托车）和车辆颜色的作业队列

最后，我们需要设计识别车辆类型和颜色的队列，并将这些统计数据汇总为最终的数据。为此，我们再次使用分片模式将作业分配到多个队列中，每个队列都有两个不同的workers：一个负责识别每辆车的位置和类型，另外一个负责识别那个位置的颜色，然后我们再次使用前面介绍的multi-worker模式将他们合并在一起。与之前一样，将代码分离到不同的容器当中可以提高颜色检测容器的重用性，可以用来检测其他事物的颜色。

这个作业队列的输出是一个JSON元组，如下所示：

   {
     "vehicles": {
        "car": 12,
        "truck": 7,
        "motorcycle": 4
      },
      "colors": {
        "white": 8,
        "black": 3,
        "blue": 6,
        "red": 6
      }
   }

该数据表示了在单个图片中找到的信息，为了将这些数据汇总在一起，我们将使用之前描述的Reduce模式，就像上面统计人口的例子中所做的，最后，这个流水线的Reduce阶段会产生汇总的结果。