Hadoop 超燃之路

作者：SoWhat1412 2021-01-26 05:15:45

以前的存儲(chǔ)手段跟分析方法現(xiàn)在行不通了!Hadoop 就是用來解決海量數(shù)據(jù)的存儲(chǔ) 跟海量數(shù)據(jù)的分析計(jì)算問題的，創(chuàng)始人 Doug Cutting 在創(chuàng)建 Hadoop 時(shí)主要思想源頭是 Google 三輛馬車.

[[378296]]

1 Hadoop 簡介

1.1 Hadoop 由來

數(shù)據(jù)容量

大數(shù)據(jù)時(shí)代數(shù)據(jù)量超級大，數(shù)據(jù)具有如下特性：

Volume(大量)
Velocity(高速)
Variety(多樣)
Value(低價(jià)值密度)

第一輛 GFS 產(chǎn)生了 HDFS。
第二輛 MapReduce 產(chǎn)生了MR。
第三輛 BigTable 產(chǎn)生了HBase。

現(xiàn)在說的 Hadoop 通常指的是 Hadoop 生態(tài)圈這樣一個(gè)廣義概念，如下：

大數(shù)據(jù)知識(shí)體系

1.2 Hadoop 特點(diǎn)

1.2.1 Hadoop 特點(diǎn)

高可用

Hadoop 底層對同一個(gè)數(shù)據(jù)維護(hù)這多個(gè)復(fù)本，即使Hadoop某個(gè)計(jì)算元素或者存儲(chǔ)出現(xiàn)問題，也不會(huì)導(dǎo)致數(shù)據(jù)的丟失。

高擴(kuò)展

在集群之間分配任務(wù)數(shù)據(jù)，可以方便的擴(kuò)展跟刪除多個(gè)節(jié)點(diǎn)，比如美團(tuán)節(jié)點(diǎn)就在3K~5k 個(gè)節(jié)點(diǎn)

高效性

在MapReduce的思想下 Hadoop是并行工作的，以加快任務(wù)的處理速度

高容錯(cuò)性

如果一個(gè)子任務(wù)速度過慢或者任務(wù)失敗 Hadoop會(huì)有響應(yīng)策略會(huì)自動(dòng)重試跟任務(wù)分配。

1.2.2 Hadoop 架構(gòu)設(shè)計(jì)

Hadoop 的 1.x 跟 2.x 區(qū)別挺大，2.x 主要是將1.x MapReduce中資源調(diào)度的任務(wù)解耦合出來交 Yarn 來管理了(接下來本文以2.7開展探索)。

1.x跟2.x變化

HDFS

Hadoop Distributed File System 簡稱 HDFS，是一個(gè)分布式文件系統(tǒng)。HDFS 有著高容錯(cuò)性，被設(shè)計(jì)用來部署在低廉的硬件上來提供高吞吐量的訪問應(yīng)用程序的數(shù)據(jù)，適合超大數(shù)據(jù)集的應(yīng)用程序。

MapReduce

MapReduce是一種編程模型，包含Map(映射) 跟 Reduce(歸約)。你可以認(rèn)為是歸并排序的深入化思想。

Yarn

Apache Hadoop YARN (Yet Another Resource Negotiator，另一種資源協(xié)調(diào)者)是一種新的 Hadoop 資源管理器，它是一個(gè)通用資源管理系統(tǒng)，可為上層應(yīng)用提供統(tǒng)一的資源管理和調(diào)度，它的引入為集群在利用率、資源統(tǒng)一管理和數(shù)據(jù)共享等方面帶來了巨大好處。

Common 組件

log組件。

獨(dú)有RPC體系ipc、I/O系統(tǒng)、序列化、壓縮。

配置文件conf。

公共方法類，比如checkSum校驗(yàn)。

2 HDFS

產(chǎn)生背景：

隨著數(shù)據(jù)量變大，數(shù)據(jù)在一個(gè)OS的磁盤無法存儲(chǔ)了，需要將數(shù)據(jù)分配到多個(gè)OS管理的磁盤中，為了方面管理多個(gè)OS下的磁盤文件，迫切需要一種系統(tǒng)來管理多臺(tái)機(jī)器上的文件，這就是分布式文件管理系統(tǒng)，HDFS 是通過目錄樹定位文件。需注意 HDFS 只是分布式文件系統(tǒng)中的其中一種。

2.1 HDFS 優(yōu)缺點(diǎn)

2.1.1 優(yōu)點(diǎn)

高容錯(cuò)性

數(shù)據(jù)會(huì)自動(dòng)保存多個(gè)副本，默認(rèn)為3個(gè)，通過增加副本來提高容錯(cuò)性。

某個(gè)副本丟失后系統(tǒng)會(huì)自動(dòng)恢復(fù)。

高擴(kuò)展性

HDFS集群規(guī)模是可以動(dòng)態(tài)伸縮的。

適合大數(shù)據(jù)處理

數(shù)據(jù)規(guī)模達(dá)到GB/TB/PB級別。

文件規(guī)模達(dá)到百萬規(guī)模以上。

流式訪問，它能保證數(shù)據(jù)的一致性。

低成本，部署廉價(jià)機(jī)器提高了商業(yè)化能了。

統(tǒng)一對外接口，Hadoop本身用Java編寫，但基于此的應(yīng)用程序可以用其他語言編寫調(diào)用。

2.1.1 缺點(diǎn)

做不到低延時(shí)

Hadoop對高吞吐做了優(yōu)化，犧牲了獲取數(shù)據(jù)的延遲，比如毫秒級獲取數(shù)據(jù)在Hadoop上做不到。

不適合存儲(chǔ)大量小文件

存儲(chǔ)大量小文件的話，它會(huì)占用 NameNode 大量的內(nèi)存來存儲(chǔ)文件、目錄和塊信息。因此該文件系統(tǒng)所能存儲(chǔ)的文件總數(shù)受限于 NameNode 的內(nèi)存容量，根據(jù)經(jīng)驗(yàn)，每個(gè)文件、目錄和數(shù)據(jù)塊的存儲(chǔ)信息大約占150字節(jié)。

小文件存儲(chǔ)的尋道時(shí)間會(huì)超過讀取時(shí)間，它違反了HDFS的設(shè)計(jì)目標(biāo)。

無法修改文件

對于上傳到HDFS上的文件，不支持修改文件，僅支持追加。HDFS適合一次寫入，多次讀取的場景。

無法并發(fā)寫入

HDFS不支持多用戶同時(shí)執(zhí)行寫操作，即同一時(shí)間，只能有一個(gè)用戶執(zhí)行寫操作。

2.2 HDFS 組成架構(gòu)

2.2.1 Client

客戶端主要有如下功能：

文件切分，文件上傳 HDFS 的時(shí)候，Client 將文件切分成一個(gè)一個(gè)的Block，然后進(jìn)行存儲(chǔ)。
與 NameNode 交互，獲取文件的位置信息。
與 DataNode 交互，讀取或者寫入數(shù)據(jù)。
Client 提供一些命令來管理 HDFS，比如啟動(dòng)或者關(guān)閉 HDFS。
Client 可以通過一些命令來訪問 HDFS。

2.2.2 NameNode

NameNode 簡稱NN，就是HDFS中的 Master，是個(gè)管理者，主要有如下功能：

管理HDFS的名稱空間。
配置副本策略
處理客戶端讀寫請求。
管理數(shù)據(jù)塊(Block)映射信息。

映射信息：NameNode(文件路徑，副本數(shù)，{Block1，Block2}，[Block1:[三個(gè)副本路徑],Block2:[三個(gè)副本路徑]])

2.2.3 DataNode

DataNode 簡稱 DN 就是HDFS集群中的Slave，NameNode 負(fù)責(zé)下達(dá)命令，DataNode執(zhí)行實(shí)際的操作。

存儲(chǔ)實(shí)際的數(shù)據(jù)塊。
執(zhí)行數(shù)據(jù)塊的讀/寫操作。

上面說過數(shù)據(jù)目錄信息存儲(chǔ)在NN中，而具體信息存儲(chǔ)在DN中，很形象的比喻如下

NN跟DN對比

DataNode 的工作機(jī)制

數(shù)據(jù)塊存儲(chǔ)在磁盤信息包括數(shù)據(jù) + 數(shù)據(jù)長度 + 校驗(yàn)和 + 時(shí)間戳。
DataNode 啟動(dòng)后向 NameNode注冊，周期性(1小時(shí))的向 NameNode 上報(bào)所有的塊信息。
NN 跟 DN 之間心跳 3秒一次，心跳返回結(jié)果帶有 NameNode 給該 DataNode 的命令如復(fù)制塊數(shù)據(jù)到另一臺(tái)機(jī)器，或刪除某個(gè)數(shù)據(jù)塊。如果超過10分鐘沒有收到某個(gè) DataNode 的心跳，則認(rèn)為該節(jié)點(diǎn)不可用。
集群運(yùn)行中可以安全加入和退出一些機(jī)器。

DataNode 確保數(shù)據(jù)完整性

當(dāng) DataNode 讀取 Block 的時(shí)候，它會(huì)計(jì)算 CheckSum。
如果計(jì)算后的 CheckSum，與 Block 創(chuàng)建時(shí)值不一樣，說明 Block 已經(jīng)損壞。
Client 讀取其他 DataNode 上的 Block。
DataNode 在其文件創(chuàng)建后周期驗(yàn)證 CheckSum

DN 進(jìn)程死亡或無法跟 NN 通信后 NN 不會(huì)立即將 DN 判死，一般經(jīng)過十分鐘 + 30秒再判刑。

2.2.4 Secondary NameNode

當(dāng) NameNode 掛掉的時(shí)候，它并不能馬上替換 NameNode 并提供服務(wù)。需要通過 HA等手段實(shí)現(xiàn)自動(dòng)切換。SNN 主要提供如下功能：

輔助 NameNode，分擔(dān)其工作量。
定期合并 Fsimage 和 Edits，并推送給 NameNode。
在緊急情況下，可輔助恢復(fù) NameNode。

2.2.5 Block

HDFS中的文件在物理上是分塊 Block 存儲(chǔ)的，在 1.x 版本中塊 = 64M，2.x中塊 = 128M。塊不是越大越好，也不是越小越好。因?yàn)橛脩臬@取數(shù)據(jù)信息時(shí)間 = 尋址塊時(shí)間 + 磁盤傳輸時(shí)間。

塊太小會(huì)增加尋址時(shí)間，程序大部分耗時(shí)在尋址上了。

快太大則會(huì)導(dǎo)致磁盤傳輸時(shí)間明顯大于尋址時(shí)間，程序處理塊數(shù)據(jù)時(shí)較慢。

2.3 HDFS 寫流程

2.3.1 具體寫流程

寫流程

客戶端通過 Distributed FileSystem 模塊向 NameNode 請求上傳文件，NameNode檢查目標(biāo)文件是否已存在，父目錄是否存在。
NameNode 返回是否可以上傳。
客戶端請求第一個(gè) Block上傳到哪幾個(gè) DataNode 服務(wù)器上。
NameNode 返回3個(gè) DataNode 節(jié)點(diǎn)，分別為dn1、dn2、dn3。
客戶端通過 FSDataOutputStream 模塊請求dn1上傳數(shù)據(jù)，dn1收到請求會(huì)繼續(xù)調(diào)用dn2，然后dn2調(diào)用dn3，將這個(gè)通信管道建立完成。
dn1、dn2、dn3逐級應(yīng)答客戶端。
客戶端開始往dn1上傳第一個(gè)Block(先從磁盤讀取數(shù)據(jù)放到一個(gè)本地內(nèi)存緩存)，以Packet為單位，dn1收到一個(gè)Packet就會(huì)傳給dn2，dn2傳給dn3;dn1每傳一個(gè)packet會(huì)放入一個(gè)應(yīng)答隊(duì)列等待應(yīng)答。
當(dāng)一個(gè) Block 傳輸完成之后，客戶端再次請求NameNode上傳第二個(gè)Block的服務(wù)器。(重復(fù)執(zhí)行3-7步)。

2.3.2 節(jié)點(diǎn)距離計(jì)算

在 HDFS 寫數(shù)據(jù)的過程中，NameNode 會(huì)選擇距離待上傳數(shù)據(jù)最近距離的DataNode接收數(shù)據(jù)。

最近距離 = 兩個(gè)節(jié)點(diǎn)到達(dá)最近的共同祖先的距離總和。

節(jié)點(diǎn)距離計(jì)算

Distance(/d1/r1/n0,/d1/r1/n0) = 0 同一節(jié)點(diǎn)上的進(jìn)程
Distance(/d1/r1/n1,/d1/r1/n2) = 2 同一機(jī)架上不同節(jié)點(diǎn)
Distance(/d1/r2/n0,/d1/r3/n2) = 4 同一數(shù)據(jù)中心不同機(jī)架節(jié)點(diǎn)
Distance(/d1/r2/n1,/d2/r4/n1) = 6 不同數(shù)據(jù)中心

2.3.3 副本節(jié)點(diǎn)選擇

第一個(gè)副本在Client所在節(jié)點(diǎn)上，如果在集群外則隨機(jī)選個(gè)。
第二個(gè)副本跟第一個(gè)副本位于同機(jī)架不同節(jié)點(diǎn)
第三個(gè)部分位于不同機(jī)架，隨機(jī)節(jié)點(diǎn)。

機(jī)架感知

2.4 HDFS 讀流程

讀流程

客戶端通過 Distributed FileSystem 向 NameNode 請求下載文件，NameNode 通過查詢元數(shù)據(jù)，找到文件塊所在的 DataNode 地址。
挑選一臺(tái) DataNode(就近原則，然后隨機(jī))服務(wù)器，請求讀取數(shù)據(jù)。
DataNode 開始傳輸數(shù)據(jù)給客戶端(從磁盤里面讀取數(shù)據(jù)輸入流，以Packet為單位來做校驗(yàn))。
客戶端以 Packet 為單位接收，先在本地緩存，然后寫入目標(biāo)文件。

2.5 NameNode 和 Secondary NameNode

2.5.1 NN 和 2NN 工作機(jī)制

NameNode 中元數(shù)據(jù)單獨(dú)存到磁盤不方便讀寫。單獨(dú)存到內(nèi)存時(shí)，斷電會(huì)丟失。Hadoop 采用的是如下方式。

FsImage :

元數(shù)據(jù)序列化后在磁盤存儲(chǔ)的地方。包含HDFS文件系統(tǒng)的所有目錄跟文件inode序列化信息。

Memory：

元數(shù)據(jù)在內(nèi)存中存儲(chǔ)的地方。

Edit 文件：

Edit 記錄客戶端更新元數(shù)據(jù)信息的每一步操作(可通過Edits運(yùn)算出元數(shù)據(jù))。

一旦元數(shù)據(jù)有更新跟添加，元數(shù)據(jù)修改追加到Edits中然后修改內(nèi)存中的元數(shù)據(jù)，這樣一旦NameNode 節(jié)點(diǎn)斷電，通過 FsImage 跟 Edits 的合并生成元數(shù)據(jù)。

Edits文件不要過大，系統(tǒng)會(huì)定期的由 Secondary Namenode 完成 FsImage 和 Edits 的合并。

NN跟2NN工作機(jī)制

第一階段：NameNode 啟動(dòng)

第一次啟動(dòng) NameNode 格式化后，創(chuàng)建 Fsimage 和 Edits 文件。如果不是第一次啟動(dòng)，直接加載編輯日志和鏡像文件到內(nèi)存。
客戶端對元數(shù)據(jù)進(jìn)行增刪改的請求。
NameNode 記錄操作日志，更新滾動(dòng)日志。
NameNode 在內(nèi)存中對數(shù)據(jù)進(jìn)行增刪改。

第二階段：Secondary NameNode 工作

Secondary NameNode 詢問 NameNode 是否需要 CheckPoint。直接帶回 NameNode 是否檢查結(jié)果。一般下面條件任意滿足即可：

CheckPoint 默認(rèn)1小時(shí)執(zhí)行一次。
一分鐘檢查一次Edits文件操作次數(shù)，達(dá)閾值 CheckPoint 。
Secondary NameNode 請求執(zhí)行 CheckPoint。
NameNode 滾動(dòng)正在寫的 Edits 日志。
將滾動(dòng)前的編輯日志Edit_001 和鏡像文件FsImage 拷貝到 Secondary NameNode。
Secondary NameNode 加載編輯日志和鏡像文件到內(nèi)存并合并。
生成新的鏡像文件 FsImage.chkpoint。
拷貝 FsImage.chkpoint 到 NameNode。
NameNode 將 FsImage.chkpoint 重新命名成 FsImage。

2.6 安全模式

NameNode 剛啟動(dòng)時(shí)候系統(tǒng)進(jìn)入安全模式(只讀)，如果整個(gè)文件系統(tǒng)中99.9%塊滿足最小副本，NameNode 會(huì)30秒后退出安全模式。

2.6.1 NameNode 啟動(dòng)

將 FsImage 文件載入內(nèi)存再執(zhí)行Edits文件各種操作，最終內(nèi)存生成完整的元數(shù)據(jù)鏡像。

創(chuàng)建個(gè)新的 FsImage 跟空 Edits 文件。

NameNode 開始監(jiān)聽 DataNode。

整個(gè)過程 NameNode 一直運(yùn)行在安全模式，NameNode 對于 Client 是只讀的。

2.6.2 DataNode 啟動(dòng)

系統(tǒng)數(shù)據(jù)塊位置不是由 NameNode 維護(hù)的，而是以塊列表形式存儲(chǔ)在 DataNode 中。

安全模式下 DataNode 向 NameNode 發(fā)送最新塊列表信息，促使 NameNode 高效運(yùn)行。

正常運(yùn)行期 NameNode 內(nèi)存中保留所有塊位置映射信息。

2.7 HDFS-HA

HDFS 集群中 NameNode存在單點(diǎn)故障(SPOF)，為了實(shí)現(xiàn) High Available，其實(shí)包括 HDFS-HA 和YARN-HA。HDFS 可以通過配置Active/Standby 兩個(gè) NameNodes 實(shí)現(xiàn)在集群中對 NameNode 的熱備來解決上述問題。如果出現(xiàn)故障，如機(jī)器崩潰或機(jī)器需要升級維護(hù)，可將NameNode很快的切換到另外一臺(tái)機(jī)器。實(shí)現(xiàn) HA 功能主要依賴ZooKeeper 跟 ZKFC 進(jìn)程。

HA故障轉(zhuǎn)移

2.7.1 HDFS-HA工作要點(diǎn)

元數(shù)據(jù)管理方式需要改變

內(nèi)存中各自保存一份元數(shù)據(jù)。

Edits 日志只有 Active 狀態(tài)的 NameNode 節(jié)點(diǎn)可以做寫操作。

兩個(gè) NameNode 都可以讀取 Edits。

共享的 Edits 放在一個(gè)共享存儲(chǔ)中管理(qjournal 或 NFS)。

需要一個(gè)狀態(tài)管理功能模塊

實(shí)現(xiàn)了一個(gè)ZKFC，常駐在每一個(gè)namenode所在的節(jié)點(diǎn)，每一個(gè)ZKFC負(fù)責(zé)監(jiān)控自己所在NameNode節(jié)點(diǎn)，利用zk進(jìn)行狀態(tài)標(biāo)識(shí)，當(dāng)需要進(jìn)行狀態(tài)切換時(shí)，由ZKFC來負(fù)責(zé)切換，切換時(shí)需要防止brain split現(xiàn)象的發(fā)生。

必須保證兩個(gè) NameNode 之間能夠ssh無密碼登錄

防腦裂，同一時(shí)刻僅僅有一個(gè) NameNode 對外提供服務(wù)。

2.7.2 ZooKeeper

ZooKeeper 提供如下功能：

故障檢測：集群中每個(gè) NameNode 在 ZooKeeper 中維護(hù)一個(gè)持久會(huì)話，如果機(jī)器崩潰，ZooKeeper中的會(huì)話將終止，ZooKeeper通知另一個(gè)NameNode需要觸發(fā)故障轉(zhuǎn)移。
現(xiàn)役NameNode選擇：ZooKeeper提供了一個(gè)簡單的機(jī)制用于唯一的選擇一個(gè)節(jié)點(diǎn)為active狀態(tài)。如果目前現(xiàn)役NameNode崩潰，另一個(gè)節(jié)點(diǎn)可能從ZooKeeper獲得特殊的排外鎖以表明它應(yīng)該成為現(xiàn)役NameNode。

2.7.3 ZKFC進(jìn)程

在 NameNode 主機(jī)上有個(gè) ZKFC(ZKFailoverController) 這樣的ZK客戶端，負(fù)責(zé)監(jiān)視管理 NameNode 狀態(tài)。ZKFC負(fù)責(zé)：

健康監(jiān)測：ZKFC周期性檢測同主機(jī)下NameNode監(jiān)控撞庫。
ZooKeeper會(huì)話管理：NameNode健康時(shí)候ZKFC保持跟ZK集群會(huì)話打開狀態(tài)，ZKFC還持有個(gè)znode鎖，如果會(huì)話終止，鎖節(jié)點(diǎn)將自動(dòng)刪除。
基于ZooKeeper的選擇：ZKFC發(fā)現(xiàn)本地NameNode健康前提下會(huì)嘗試獲取znode鎖，獲得成功則Active狀態(tài)。

3 MapReduce

MapReduce是個(gè)分布式運(yùn)算程序的編程框架，是基于 Hadoop 的數(shù)據(jù)分析計(jì)算核心框架。處理過程分為兩個(gè)階段：Map 階段跟 Reduce 階段。

Map 負(fù)責(zé)把一個(gè)任務(wù)分解成多個(gè)任務(wù)。該階段的 MapTask 并發(fā)實(shí)例，完全并行運(yùn)行，互不相干。

Reduce 負(fù)責(zé)把多個(gè)任務(wù)處理結(jié)果匯總。該階段的 ReduceTask 并發(fā)實(shí)例互不相干，但是他們的數(shù)據(jù)依賴于上一個(gè)階段的所有 MapTask 并發(fā)實(shí)例的輸出。

MapReduce 編程模型只能包含一個(gè) Map 階段和一個(gè) Reduce 階段，如果用戶的業(yè)務(wù)邏輯非常復(fù)雜，那就只能多個(gè)MapReduce程序串行運(yùn)行。

用戶編寫MR任務(wù)時(shí)候程序?qū)崿F(xiàn)部分分成三個(gè)部分：Mapper、Reducer、Driver(提交運(yùn)行mr程序的客戶端)。

3.1 優(yōu)缺點(diǎn)

3.1.1 優(yōu)點(diǎn)

易于編程

簡單實(shí)現(xiàn)了一些接口就可以完成個(gè)分布式程序，你寫個(gè)分布式程序跟寫個(gè)串行化程序一樣，類似八股文編程。

良好的擴(kuò)展

計(jì)算資源不足時(shí)可以簡單的增加機(jī)器來擴(kuò)展計(jì)算能力。

高容錯(cuò)性

MapReduce任務(wù)部署在多臺(tái)機(jī)器上后如果其中一臺(tái)掛了，系統(tǒng)會(huì)進(jìn)行自動(dòng)化的任務(wù)轉(zhuǎn)移來保證任務(wù)正確執(zhí)行。

適合PB級數(shù)據(jù)離線處理

比如美團(tuán)3K個(gè)節(jié)點(diǎn)的集群并發(fā)，提供超大數(shù)據(jù)處理能力。

3.1.2 缺點(diǎn)

不擅長實(shí)時(shí)計(jì)算

MapReduce 不會(huì)想 MySQL 一樣毫秒級返回結(jié)果。

不擅長流式計(jì)算

流式計(jì)算的輸入數(shù)據(jù)是動(dòng)態(tài)的，而 MapReduce 的輸入數(shù)據(jù)集是靜態(tài)的。

不擅長DAG計(jì)算

多個(gè)應(yīng)用程序存在依賴關(guān)系，MapReduce的作業(yè)結(jié)果會(huì)落盤導(dǎo)致大量磁盤IO，性能賊低，此時(shí)上Spark吧!

3.2 序列化

序列化

把內(nèi)存中的對象，轉(zhuǎn)換成字節(jié)序列(或其他數(shù)據(jù)傳輸協(xié)議)以便于存儲(chǔ)(持久化)和網(wǎng)絡(luò)傳輸。

反序列化

將收到字節(jié)序列(或其他數(shù)據(jù)傳輸協(xié)議)或者是硬盤的持久化數(shù)據(jù)，轉(zhuǎn)換成內(nèi)存中的對象。

因?yàn)?Hadoop 在集群之間進(jìn)行通訊或者 RPC 調(diào)用時(shí)是需要序列化的，而且要求序列化要快、且體積要小、占用帶寬要小。而Java自帶的序列化是重量級框架，對象序列化后會(huì)附帶額外信息，比如各種校驗(yàn)信息，header，繼承體系等。所以 Hadoop 自研了序列化框架。

Java類型	Hadoop Writable類型
boolean	BooleanWritable
byte	ByteWritable
int	IntWritable
float	FloatWritable
long	LongWritable
double	DoubleWritable
String	Text
map	MapWritable
array	ArrayWritable

3.3 MapTask 并行度

數(shù)據(jù)塊：Block 是 HDFS 物理上把數(shù)據(jù)分成一塊一塊。

數(shù)據(jù)切片：數(shù)據(jù)切片只是在邏輯上對輸入進(jìn)行分片，并不會(huì)在磁盤上將其切分成片進(jìn)行存儲(chǔ)。

切片核心注意點(diǎn)：

一個(gè) Job 的 Map 階段并行度又客戶端提交Job時(shí)的切片數(shù)決定
每個(gè) Split 切片分配個(gè) MapTask 并行實(shí)例處理
模型情況下切片大小 = BlockSize
切片時(shí)不會(huì)考慮數(shù)據(jù)集整體大小，而是逐個(gè)針對每個(gè)文件單獨(dú)切片的。

3.3.1 FileInputFormat 切片源碼追蹤

FileInputFormat切片源碼追蹤
程序先找到目標(biāo)數(shù)據(jù)存儲(chǔ)目錄
開始遍歷目錄下每個(gè)文件。每個(gè)文件都會(huì)做如下操作
獲取切片大小，默認(rèn)情況下切片大小 = blocksize
開始切片，每次切片都要判斷剩余部分是否大于塊的1.1倍，不大于則就劃分到一個(gè)切片。
切片信息寫到切片規(guī)劃文件中。
切片核心過程在getSplit方法完成。
InputSplit只是記錄了切片元數(shù)據(jù)信息，如起始位置、長度跟所在節(jié)點(diǎn)列表等。

3.3.2 切片大小計(jì)算

SplitSize= Math.max(minSize,Math.min(maxSize,blockSize))

mapreduce.input.fileinputformat.split.minsize 默認(rèn) 1
mapreduce.input.fileinputformat.split.maxsize 默認(rèn) Long.MAXValue
blockSize 默認(rèn)128M
maxsize ：該參數(shù)如果比blockSize小灰導(dǎo)致切片變小，且就等于配置的整個(gè)參數(shù)。
minsize ：該參數(shù)如果調(diào)的比blockSize大，則切片大小會(huì)比blockSize還大。

3.3.3 切片舉例

切片舉例

3.4 FileInputFormat

3.4.1 實(shí)現(xiàn)類簡介

MR任務(wù)輸入文件個(gè)數(shù)各有不同，針對不同類型MR定義了一個(gè)接口跟若干實(shí)現(xiàn)類來讀取不同的數(shù)據(jù)。

input繼承關(guān)系

TextInputFormat

默認(rèn)使用類，按行讀取每條數(shù)據(jù)，Key是該行數(shù)據(jù)的 offset，Value = 行內(nèi)容。

KeyValueTExtInputFormat

每行都是一條記錄，被指定分隔符分割為Key跟Value，默認(rèn)是 \t 。

NLineInputFormat

該模型下每個(gè) map 處理 InputSplit 時(shí)不再按照 Block 塊去劃分，而是按照指定的行數(shù)N來劃分文件。

自定義InputFormat

基礎(chǔ)接口，改寫 RecordReader，實(shí)現(xiàn)一次讀取一個(gè)完整文件封裝為 KV，使用 SequenceFileOutPutFormat 輸出合并文件。

CombineTextInputFormat

用于小文件過多場景，邏輯上合并多個(gè)小文件個(gè)一個(gè)切片任務(wù)。較重要中

3.4.2 CombineTextInputFormat

默認(rèn)框架 TextInputFormat 切片機(jī)制是對任務(wù)按文件規(guī)劃切片，不管文件多小，都會(huì)是一個(gè)單獨(dú)的切片，都會(huì)交給一個(gè)MapTask，這樣如果有大量小文件，就會(huì)產(chǎn)生大量的MapTask，處理效率極其低下。CombineTextInputFormat 可以將多個(gè)小文件從邏輯上規(guī)劃到一個(gè)切片中，這樣多個(gè)小文件就可以交給一個(gè)MapTask處理。主要包含虛擬存儲(chǔ)過程跟切片過程。

CombineTextInputFormat.setMaxInputSplitSize(job, 4194304); // 4m

虛擬存儲(chǔ)過程：

文件 <= SplitSize 則單獨(dú)一塊。
1 * SplitSize < 文件 < 2 * SplitSize 時(shí)對半分。
文件 >= 2*SplitSize時(shí)，以 SplitSize 切割一塊，剩余部分若 < 2 * SplitSize 則對半分。

切片過程：

判斷虛擬存儲(chǔ)的文件大小是否大于setMaxInputSplitSize值，大于等于則單獨(dú)形成一個(gè)切片。

如果不大于則跟下一個(gè)虛擬存儲(chǔ)文件進(jìn)行合并，共同形成一個(gè)切片。

切片過程

3.6 OutputFormat

OutputFormat 是 MapReduce 輸出的基類，常見的實(shí)現(xiàn)類如下：

3.5.1 TextOutputFormat

系統(tǒng)默認(rèn)輸出格式，把每條記錄寫為文本行，他的K跟V是任意類型，系統(tǒng)在寫入時(shí)候會(huì)統(tǒng)一轉(zhuǎn)化為字符串。

3.5.2 SequenceFileOutputFormat

此模式下的輸出結(jié)果作為后續(xù)MapReduce任務(wù)的輸入，該模式下數(shù)據(jù)格式緊湊，很容易被壓縮。

3.5.3 自定義OutputFormat

如果需求不滿足可按需求進(jìn)行自定義。

自定義類繼承自FileOutputFormat。
重寫RecordWriter，改寫具體輸出數(shù)據(jù)的方法write。

3.6 MapReduce 流程

3.6.1 整體流程圖

MapReduce流程

MapTask 工作機(jī)制

Read階段：MapTask 通過用戶編寫的RecordReader，從輸入InputSplit中解析出一個(gè)個(gè)key/value。
Map階段：將解析出的key/value交給用戶編寫map()函數(shù)處理，并產(chǎn)生一系列新的key/value。
Collect收集階段：它會(huì)將生成的key/value分區(qū)(調(diào)用Partitioner)，并寫入一個(gè)環(huán)形內(nèi)存緩沖區(qū)中。
Spill階段：先按照分區(qū)進(jìn)行排序，然后區(qū)內(nèi)按照字典對key進(jìn)行快排，并在必要時(shí)對數(shù)據(jù)進(jìn)行合并、壓縮等操作。
Combine階段：選擇性可進(jìn)行MapTask內(nèi)的優(yōu)化提速。

ReduceTask 工作機(jī)制

Copy階段：從所有的MapTask中收集結(jié)果然后決定將數(shù)據(jù)放入緩存還是磁盤。
Merge階段：copy數(shù)據(jù)時(shí)后天會(huì)對磁盤還有內(nèi)存數(shù)據(jù)進(jìn)行Merge。
Sort階段：ReduceTask需對所有數(shù)據(jù)進(jìn)行一次歸并排序，方便執(zhí)行reduce 函數(shù)。
Reduce階段：調(diào)用用戶 reduce() 函數(shù)將計(jì)算結(jié)果寫到HDFS上。

3.6.2 Shuffle

Shuffle機(jī)制

MapReduce 的核心就是 Shuffle 過程，Shuffle 過程是貫穿于 map 和 reduce 兩個(gè)過程的!在Map端包括Spill過程，在Reduce端包括copy和sort過程。具體Shuffle過程如下：

MapTask 收集我們的map()方法輸出的kv對，放到內(nèi)存緩沖區(qū)中。
從內(nèi)存緩沖區(qū)不斷溢出本地磁盤文件，可能會(huì)溢出多個(gè)文件，溢出前會(huì)按照分區(qū)針對key進(jìn)行區(qū)內(nèi)快排。
多個(gè)溢出文件會(huì)被合并成大的溢出文件。
在溢出過程及合并的過程中，都要調(diào)用 Partitioner 進(jìn)行分區(qū)和針對key進(jìn)行排序。
ReduceTask 根據(jù)自己的分區(qū)號，去各個(gè) MapTask 機(jī)器上取相應(yīng)的結(jié)果分區(qū)數(shù)據(jù)。
ReduceTask 對收集后的數(shù)據(jù)進(jìn)行合并跟歸并排序。
進(jìn)入 ReduceTask 的邏輯運(yùn)算過程，調(diào)用用戶自定義的reduce()方法。
Shuffle 中的緩沖區(qū)大小會(huì)影響到 MapReduce 程序的執(zhí)行效率，原則上說，緩沖區(qū)越大，磁盤io的次數(shù)越少，執(zhí)行速度就越快。

3.6.3 Partition

MapReduce 默認(rèn)的分區(qū)方式是hashPartition，在這種分區(qū)方式下，KV 對根據(jù) key 的 hashcode 值與reduceTask個(gè)數(shù)進(jìn)行取模，決定該鍵值對該要訪問哪個(gè)ReduceTask。

public int getPartition(K2 key, V2 value, int numReduceTasks) { 
    return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks; 
    // numReduceTasks 默認(rèn) = 1 所以導(dǎo)致默認(rèn)的reduce結(jié)果 = 1 
  }

自定義的時(shí)候一般就是類繼承Partitioner然后重寫getPartition 方法。用戶也可以設(shè)置ReduceTask數(shù)量，不過會(huì)遵循如下規(guī)則。

如果 ReduceTask 數(shù) > getPartition 數(shù)，會(huì)多產(chǎn)生幾個(gè)空的輸出part-r-000xx。
如果 1 < ReduceTask < getPartition 數(shù)，會(huì)有部分?jǐn)?shù)據(jù)無法安放導(dǎo)致報(bào)錯(cuò)。
如果ReduceTask = 1，不管MapTask端輸出多少分區(qū)文件結(jié)果都是一個(gè)文件。
分區(qū)必須從0開始，逐步累加。

比如假設(shè)自定義分區(qū)數(shù)為5。

job.setNumReduceTasks(1)：會(huì)正常運(yùn)行，只不過會(huì)產(chǎn)生一個(gè)輸出文件。
job.setNumReduceTasks(2)：會(huì)報(bào)錯(cuò)。
job.setNumReduceTasks(6)：大于5，程序會(huì)正常運(yùn)行，會(huì)產(chǎn)生空文件。

3.6.4 環(huán)形緩沖區(qū)

Map 的輸出結(jié)果由 Collector 處理，每個(gè) Map 任務(wù)不斷地將鍵值對輸出到在內(nèi)存中構(gòu)造的一個(gè)環(huán)形數(shù)據(jù)結(jié)構(gòu)中。使用環(huán)形數(shù)據(jù)結(jié)構(gòu)是為了更有效地使用內(nèi)存空間，在內(nèi)存中放置盡可能多的數(shù)據(jù)。

環(huán)形數(shù)據(jù)結(jié)構(gòu)其實(shí)就是個(gè)字節(jié)數(shù)組byte[]，叫kvbuffer，默認(rèn)值100M。里面主要存儲(chǔ) 數(shù)據(jù) 跟元數(shù)據(jù)。中間有個(gè)分界點(diǎn)，并且分界點(diǎn)是變化的。當(dāng)環(huán)形緩沖區(qū)寫入的buffer的大小達(dá)到 80% 滿足溢寫條件的時(shí)候，開始溢寫spill。系統(tǒng)有兩個(gè)線程一個(gè)負(fù)責(zé)寫入數(shù)據(jù)，一個(gè)負(fù)責(zé)spill數(shù)據(jù)。

數(shù)據(jù)：

存儲(chǔ) Key + Value + bufindex。其中 bufindex(即數(shù)據(jù)的存儲(chǔ)方向)是一直悶著頭地向上增長，比如bufindex初始值為0，一個(gè)Int型的key寫完之后，bufindex增長為4，一個(gè)Int型的value寫完之后，bufindex增長為8。

元數(shù)據(jù)：

元數(shù)據(jù)是為了排序而生，是關(guān)于數(shù)據(jù)描述的數(shù)據(jù)。

Kvmeta = Partition + keystart + valstart + valLength ，共占用4個(gè)Int長度，其中K的長度 = V的起點(diǎn) - K的起點(diǎn)。

Kvmeta 的存放指針 Kvindex 每次都是向下跳四個(gè) 格子，然后再向上一個(gè)格子一個(gè)格子地填充四元組的數(shù)據(jù)。比如Kvindex初始位置是-4，當(dāng)?shù)谝粋€(gè)鍵值對寫完之后，(Kvindex+0)的位置存放partition的起始位置、(Kvindex+1)的位置存放keystart、(Kvindex+2)的位置存放valstart、(Kvindex+3)的位置存放value length，然后Kvindex跳到 -8位置，等第二個(gè)鍵值對和索引寫完之后，Kvindex跳到-12位置。

kvmeta.put(kvindex + PARTITION, partition); 
2kvmeta.put(kvindex + KEYSTART, keystart); 
3kvmeta.put(kvindex + VALSTART, valstart); 
4kvmeta.put(kvindex + VALLEN, distanceTo(valstart, valend)); 
5// advance kvindex 改變每次index的值 每次4個(gè)位置! 
6kvindex = (kvindex - NMETA + kvmeta.capacity()) % kvmeta.capacity();

環(huán)形緩沖區(qū)

3.6.5 Combiner 合并

Combiner 是 MR 程序中 Mapper 跟 Reducer 之外的組件。
Combiner 是在每一個(gè)MapTask 所在節(jié)點(diǎn)運(yùn)行，Reducer 是接受全部 Mapper 輸出結(jié)果。
Combiner 屬于局部匯總的意思，來減少網(wǎng)絡(luò)傳輸。
Combiner 用的時(shí)候要注意不能影響最終業(yè)務(wù)邏輯!比如求平均值就不能用。求和就OK。

3.6.6 關(guān)于 MapReduce 排序

MapReduce框架最重要的操作就是排序，MapTask 跟 ReduceTask 都會(huì)根據(jù)key進(jìn)行按照字典順序進(jìn)行快排。

MapTask 將緩沖區(qū)數(shù)據(jù)快排后寫入到磁盤，然后磁盤文件會(huì)進(jìn)行歸并排序。

ReduceTask統(tǒng)一對內(nèi)存跟磁盤所有數(shù)據(jù)進(jìn)行歸并排序。

3.6.7 ReduceJoin 跟 MapJoin

Reducejoin

思路：通過將關(guān)聯(lián)條件作為Map 輸出的 Key，將兩表滿足 Join條件的數(shù)據(jù)并攜帶數(shù)據(jù)源文件發(fā)送同一個(gè)ReduceTask，在Reduce端進(jìn)行數(shù)據(jù)串聯(lián)信息合并。

缺點(diǎn)：合并操作在Reduce端完成，Reduce 端處理壓力太大，并且Reduce端易產(chǎn)生數(shù)據(jù)傾斜。

MapJoin

適用：適用于一張表十分小、一張表很大的場景。

思路：在 Map 端緩存多張表，提前處理業(yè)務(wù)邏輯，這樣增加 Map 端業(yè)務(wù)，減少 Reduce 端數(shù)據(jù)的壓力，盡可能的減少數(shù)據(jù)傾斜。

3.6.8 注意點(diǎn)

ReduceTask = 0 說明沒有Reduce節(jié)點(diǎn)，輸出文件個(gè)數(shù)和 Map 個(gè)數(shù)一樣。

ReduceTask 默認(rèn)= 1，所以結(jié)果是一個(gè)文件。

ReduceTask 的個(gè)數(shù)不是任意設(shè)置的，需跟集群性能還有結(jié)果需求而定。

邏輯處理 Mapper 時(shí)候可根據(jù)業(yè)務(wù)需求實(shí)現(xiàn)其中三個(gè)方法，map、setup、cleanup。

3.7 壓縮

壓縮是提高Hadoop運(yùn)行效率的一種優(yōu)化策略，通過在Mapper、Reducer運(yùn)行過程的數(shù)據(jù)進(jìn)行壓縮來減少磁盤空間跟網(wǎng)絡(luò)傳輸，最終實(shí)現(xiàn)提高M(jìn)R運(yùn)行速度。但需注意壓縮也給CPU運(yùn)算帶來了負(fù)擔(dān)。

壓縮的基本原則：

運(yùn)算密集型任務(wù) ，少壓縮。

IO密集型任務(wù)，多壓縮。

壓縮格式	自帶	算法	擴(kuò)展名	可切分嗎	壓縮后，代碼修改
DEFLATE	是	DEFLATE	.deflate	否	不需要修改
Gzip	是	DEFLATE	.gz	否	不需要修改
bzip2	是	bzip2	.bz2	是	不需要修改
Snappy	否	Snappy	.snappy	否	不需要修改
LZO	否	LZO	.lzo	是	需要建索引還需要指定輸入格式

4 YARN

Yarn 是一個(gè)資源調(diào)度平臺(tái)，負(fù)責(zé)為運(yùn)算程序提供服務(wù)器運(yùn)算資源，相當(dāng)于一個(gè)分布式的操作系統(tǒng)平臺(tái)，而 MapReduce 等運(yùn)算程序則相當(dāng)于運(yùn)行于操作系統(tǒng)之上的應(yīng)用程序。

4.1 基本組成

Yarn架構(gòu)

YARN主要由 ResourceManager、NodeManager、ApplicationMaster 和 Container 等組件構(gòu)成。

ResourceManager

處理客戶端請求

監(jiān)控NodeMananger

啟動(dòng)或監(jiān)控ApplicationMaster

計(jì)算資源的分配跟調(diào)度

NodeManager

管理單個(gè)節(jié)點(diǎn)上資源

處理來著ResourceManager的命令

處理來自ApplicationMaster的命令

ApplicationMaster

負(fù)責(zé)數(shù)據(jù)切分。

為應(yīng)用程序申請資源并分配給內(nèi)部任務(wù)。

任務(wù)的監(jiān)控跟容錯(cuò)。

Container

Container 是 YARN 中資源的抽象，封裝了某個(gè)節(jié)點(diǎn)上的多維度資源，比如內(nèi)存、CPU、磁盤、網(wǎng)絡(luò)等。

YarnChild 其實(shí)它就是一個(gè)運(yùn)行程序的進(jìn)程。MrAppMaster 運(yùn)行程序時(shí)向 Resouce Manager 請求的 Maptask / ReduceTask。

4.2 Yarn 調(diào)度 MapReduce 任務(wù)

Yarn調(diào)度流程

當(dāng) MR 程序提交到客戶端所在的節(jié)點(diǎn)時(shí)后大致運(yùn)行流程如下：

作業(yè)提交

Client 調(diào)用 job.waitForCompletion 方法 YarnRunner ，向整個(gè)集群提交MapReduce作業(yè)。Client 向 RM 申請一個(gè)作業(yè)id。

RM 給 Client 返回該 job 資源的提交路徑和作業(yè) id。

Client 提交jar包、切片信息和配置文件到指定的資源提交路徑。

Client 提交完資源后，向 RM 申請運(yùn)行 MrAppMaster。

作業(yè)初始化

當(dāng) RM 收到 Client 的請求后，將該 Task 添加到容量調(diào)度器中。

某一個(gè)空閑的 NodeManager 領(lǐng)取到該 Task 。

該 NodeManager 創(chuàng)建 Container，并產(chǎn)生 MRAppMaster。

下載 Client 提交的資源到本地。

任務(wù)分配

MRAppMaster 向 RM 申請運(yùn)行多個(gè) MapTask 任務(wù)資源。

RM 將運(yùn)行 MapTask 任務(wù)分配給倆 NodeManager。其中分配原則是優(yōu)先 jar 跟數(shù)據(jù)在一臺(tái)機(jī)器上，其次就盡可能在一個(gè)機(jī)房。最后隨便來個(gè)空閑機(jī)器。

任務(wù)運(yùn)行

MR 向兩個(gè)接收到任務(wù)的 NodeManager 發(fā)送程序啟動(dòng)腳本，這兩個(gè) NodeManager 分別啟動(dòng)MapTask，MapTask 對數(shù)據(jù)分區(qū)排序。

MrAppMaster 等待所有 MapTask 運(yùn)行完畢后，向RM申請容器運(yùn)行ReduceTask。

ReduceTask 向 MapTask 獲取相應(yīng)分區(qū)的數(shù)據(jù)。

程序運(yùn)行完畢后，MR會(huì)向RM申請注銷自己。

進(jìn)度和狀態(tài)更新

YARN 中的任務(wù)將其進(jìn)度和狀態(tài)(包括counter)返回給應(yīng)用管理器，客戶端每秒向應(yīng)用管理器請求進(jìn)度更新來展示給用戶。

作業(yè)完成

除了向應(yīng)用管理器請求作業(yè)進(jìn)度外, 客戶端每5秒都會(huì)通過調(diào)用 waitForCompletion() 來檢查作業(yè)是否完成。作業(yè)完成之后, 應(yīng)用管理器和Container會(huì)清理工作狀態(tài)。作業(yè)的信息會(huì)被作業(yè)歷史服務(wù)器存儲(chǔ)以備之后用戶核查。

4.3 資源調(diào)度器

目前，Hadoop作業(yè)調(diào)度器主要有三種：FIFO、Capacity Scheduler 和 Fair Scheduler。Hadoop2.7.2默認(rèn)的資源調(diào)度器是Capacity Scheduler。

4.3.1 FIFO

FIFO調(diào)度

4.3.2 容量調(diào)度器 Capacity Scheduler

容量調(diào)度器

支持多個(gè)隊(duì)列，每個(gè)隊(duì)列配置一定資源，每個(gè)隊(duì)列采用FIFO策略。
為防止同一個(gè)童虎作業(yè)獨(dú)占隊(duì)列資源，會(huì)對同一用戶提交作業(yè)所占資源量限制。
計(jì)算每個(gè)隊(duì)列中在跑任務(wù)數(shù)與其應(yīng)該分得的計(jì)算只有比值，選擇個(gè)比值最小的隊(duì)列(最閑的)。
按照作業(yè)優(yōu)先級跟提交時(shí)間，同時(shí)還考慮用戶資源限制跟內(nèi)存限制對隊(duì)列任務(wù)排序。
比如job1、job2、job3分配排在最前面也是并行運(yùn)行。

4.3.3 公平調(diào)度器 Fair Scheduler

支持多隊(duì)列多用戶，每個(gè)隊(duì)列中資源可以配置，同一隊(duì)列中作業(yè)公平共享隊(duì)列中所有資源。

公平調(diào)度器

比如有queue1、queue2、queue3三個(gè)任務(wù)隊(duì)列，每個(gè)隊(duì)列中的job按照優(yōu)先級分配資源，優(yōu)先級高獲得資源多，但會(huì)確保每個(gè)任務(wù)被分配到資源。

每個(gè)任務(wù)理想所需資源跟實(shí)際獲得資源的差距叫缺額，同一個(gè)隊(duì)列中是按照缺額高低來先后執(zhí)行的，缺額越大越優(yōu)先獲得資源。

4.4 任務(wù)推測執(zhí)行

作業(yè)完成時(shí)間取決于最慢的任務(wù)完成時(shí)間。系統(tǒng)中有99%的Map任務(wù)都完成了，只有少數(shù)幾個(gè)Map老是進(jìn)度很慢，此時(shí)系統(tǒng)會(huì)發(fā)現(xiàn)拖后腿的任務(wù)，比如某個(gè)任務(wù)運(yùn)行速度遠(yuǎn)慢于任務(wù)平均速度。為拖后腿任務(wù)啟動(dòng)一個(gè)備份任務(wù)，同時(shí)運(yùn)行。誰先運(yùn)行完，則采用誰的結(jié)果。

5 MapReduce 優(yōu)化方法

MapReduce優(yōu)化方法主要從六個(gè)方面考慮：數(shù)據(jù)輸入、Map階段、Reduce階段、IO傳輸、數(shù)據(jù)傾斜問題和常用的調(diào)優(yōu)參數(shù)。

5.1 數(shù)據(jù)輸入

數(shù)據(jù)采集時(shí)，用 Hadoop Archive 將多個(gè)小文件打包成一個(gè)Har文件。

業(yè)務(wù)處理前，SequenceFile 由一系列KV組成，key=文件名，value=文件內(nèi)容，將大批小文件合并成大文件。

在 MapReduce 處理時(shí)，采用CombineTextInputFormat來作為輸入，解決輸入端大量小文件場景。

對于大量小文件任務(wù)開啟JVM 重用可提速，JVM 重用可以使得 JVM 實(shí)例在同一個(gè) job 中重新使用N次。N的值可以在Hadoop的mapred-site.xml文件中進(jìn)行配置，通常在10-20之間。

5.2 Map 階段

減少溢寫 Spill 次數(shù)，調(diào)整循環(huán)緩存區(qū)大小，減少磁盤IO。

減少合并 Merge 次數(shù)，增大Merge文件大小減少次數(shù)。

在不影響業(yè)務(wù)的情況下在Map端進(jìn)行Combine處理。

5.3 Reduce 階段