前面的章節描述了一系列將佇列拆分和連線在一起以實現更復雜批處理的模式，複製和生成多個不同的輸出是批處理的重要組成部分，但有時將多個輸出合併到一起以生成某種聚合輸出也同樣很重要，如圖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階段會產生彙總的結果。