批處理工作流的輸出

我們已經(jīng)討論了串起 MapReduce 工作流的一些算法，但我們忽略了一個(gè)重要的問題：當(dāng)工作流結(jié)束后，處理結(jié)果是什么？我們一開始是為什么要跑這些任務(wù)來著？

對(duì)于數(shù)據(jù)庫查詢場景，我們會(huì)區(qū)分事務(wù)型處理場景（OLTP）和分析性場景（OLAP，參見事務(wù)型還是分析型）。我們觀察到，OLTP 場景下的查詢通常只會(huì)涉及很小的一個(gè)數(shù)據(jù)子集，因此通常會(huì)使用索引加速查詢，然后將結(jié)果展示給用戶（例如，使用網(wǎng)頁展示）。另一方面，分析型查詢通常會(huì)掃描大量的數(shù)據(jù)記錄，執(zhí)行分組（grouping）和聚集（aggregating）等統(tǒng)計(jì)操作，然后以報(bào)表的形式呈現(xiàn)給用戶：比如某個(gè)指標(biāo)隨時(shí)間的變化曲線、依據(jù)某種排序方式的前十個(gè)數(shù)據(jù)條目、將數(shù)據(jù)按子類分解并統(tǒng)計(jì)其分布。這些報(bào)表通常會(huì)用于輔助分析員或者經(jīng)理進(jìn)行商業(yè)決策。

那批處理處于一個(gè)什么位置呢？它既不是事務(wù)型，也不是分析型。當(dāng)讓，從輸入數(shù)據(jù)量的角度來說，批處理更接近分析型任務(wù)。然而，一組 MapReduce 任務(wù)組成的執(zhí)行流通常和用于分析型的 SQL 查詢并不相同（參見 Hadoop 和分布式數(shù)據(jù)庫的對(duì)比）。批處理的輸出通常不是一個(gè)報(bào)表，而是另外某種格式的數(shù)據(jù)。

構(gòu)建查詢索引

谷歌發(fā)明 MapReduce 大數(shù)據(jù)處理框架的最初動(dòng)機(jī)就是解決搜索引擎的索引問題，開始時(shí)通過 5~10 個(gè) MapReduce 工作流來為搜索引擎來構(gòu)建索引。盡管谷歌后面將 MapReduce 使用拓展到了其他場景，仔細(xì)考察構(gòu)建搜索引擎索引的過程，有助于深入地了解 MapReduce（當(dāng)然，即使到今天， Hadoop MapReduce 仍不失為一個(gè)給 Lucene/Solr 構(gòu)建索引的好辦法）。

我們?cè)凇叭乃饕湍：饕币还?jié)粗策略的探討過像 Lucene 這樣的全文索引引擎是如何工作的：倒排索引是一個(gè)詞表（the term dictionary），利用該詞表，你可以針對(duì)關(guān)鍵詞快速地查出對(duì)應(yīng)文檔列表（the postings list）。當(dāng)然，這是一個(gè)很簡化的理解，在實(shí)踐中，索引還需要很多其他信息，包括相關(guān)度，拼寫訂正，同義詞合并等等，但其背后的原理是不變的。

如果你想在一個(gè)固定文檔集合上構(gòu)建全文索引，批處理非常合適且高效：

1. Mapper 會(huì)將文檔集合按合適的方式進(jìn)行分區(qū)
2. Reducer 會(huì)對(duì)每個(gè)分區(qū)構(gòu)建索引
3. 最終將索引文件寫回分布式文件系統(tǒng)

構(gòu)建這種按文檔分區(qū)（document-partitioned，與 term-partitioned 相對(duì)，參見分片和次級(jí)索引）的索引，可以很好地并發(fā)生成。由于使用關(guān)鍵詞進(jìn)行索引查詢是一種只讀操作，因此，這些索引文件一旦構(gòu)建完成，就是不可變的（immutable）。

如果被索引的文檔集發(fā)生變動(dòng)，一種應(yīng)對(duì)策略是，定期針對(duì)所有文檔重跑全量索引構(gòu)建工作流（workflow），并在索引構(gòu)建完時(shí)使用新的索引對(duì)舊的進(jìn)行整體替換。如果兩次構(gòu)建之間，僅有一小部分文檔發(fā)生了變動(dòng)，則這種方法代價(jià)實(shí)在有點(diǎn)高。但也有優(yōu)點(diǎn)，索引構(gòu)建過程很好理解：文檔進(jìn)去，索引出來。

當(dāng)然，我們也可以增量式的構(gòu)建索引。我們?cè)诘谌掠懻撨^，如果你想增加、刪除或者更新文檔集，Lucene 就會(huì)構(gòu)建新的索引片段，并且異步地將其與原有索引進(jìn)行歸并（merge）和壓實(shí)（compact）。我們將會(huì)在第十一章就增量更新進(jìn)行更深入的討論。

以 KV 存儲(chǔ)承接批處理輸出

搜索索引只是批處理工作流一種可能的輸出。批處理其他的用途還包括構(gòu)建機(jī)器學(xué)習(xí)系統(tǒng)，如分類器（classifiers，如 垃圾郵件過濾，同義詞檢測(cè)，圖片識(shí)別）和推薦系統(tǒng)（recommendation system，如你可能認(rèn)識(shí)的人，可能感興趣的產(chǎn)品或者相關(guān)的檢索）。

這些批處理任務(wù)的輸出通常在某種程度是數(shù)據(jù)庫：如，一個(gè)可以通過用戶 ID 來查詢其可能認(rèn)識(shí)的人列表的數(shù)據(jù)庫，或者一個(gè)可以通過產(chǎn)品 ID 來查詢相關(guān)產(chǎn)品的數(shù)據(jù)庫。

web 應(yīng)用會(huì)查詢這些數(shù)據(jù)庫來處理用戶請(qǐng)求，這些應(yīng)用通常不會(huì)跟 Hadooop 生態(tài)部署在一塊。那么，如何讓批處理的輸出寫回?cái)?shù)據(jù)庫，以應(yīng)對(duì) web 應(yīng)用的查詢？

最直觀的做法是，在 Mapper 或者 Reducer 代碼邏輯中，使用相關(guān)數(shù)據(jù)庫的客戶端庫，將 Mapper 或者 Reducer 的輸出直接寫入數(shù)據(jù)庫服務(wù)器，每次一個(gè)記錄。這種方式能夠工作（須假設(shè)防火墻允許我們直接從 Hadoop 中訪問生產(chǎn)環(huán)境的數(shù)據(jù)庫服務(wù)器），但往往并不是一個(gè)好的做法：

• 吞吐不匹配。正如之前 join 一節(jié)中所討論的，通過網(wǎng)絡(luò)一條條寫入記錄的吞吐要遠(yuǎn)小于一個(gè)批處理任務(wù)的吞吐。即使數(shù)據(jù)庫的客戶端通常支持將多個(gè) record 寫入 batch 成一個(gè)請(qǐng)求，性能仍然會(huì)比較差。
• 數(shù)據(jù)庫過載。一個(gè) MapReduce 任務(wù)通常會(huì)并行地跑很多個(gè)子任務(wù)。如果所有 Mapper 和 Reducer ，以批處理產(chǎn)生輸出的速率，并發(fā)地將輸出寫到同一個(gè)數(shù)據(jù)庫，則該數(shù)據(jù)庫很容會(huì)被打爆（overwhelmed）。此時(shí)，其查詢性能大概率非常差，也因此難以對(duì)外提供正常服務(wù)，從而給系統(tǒng)的其他組件帶來運(yùn)維問題。
• 可能產(chǎn)生副作用。通常來說，MapReduce 對(duì)外提供簡單的“全有或全無（all-or-nothing）”的輸出保證：如果整個(gè)任務(wù)成功，即使子任務(wù)一時(shí)失敗重試，但最終的輸出也會(huì)看起來像運(yùn)行了一次；如果整個(gè)任務(wù)失敗，則沒有任何輸出。但直接從任務(wù)內(nèi)部將輸出寫入外部服務(wù)，會(huì)產(chǎn)生外部可見的副作用。在這種情況下，你就必須考慮任務(wù)的部分成功狀態(tài)可能會(huì)暴露給其他系統(tǒng)，并要理解 Hadoop 內(nèi)部重試和推測(cè)執(zhí)行的復(fù)雜機(jī)制。

一個(gè)更好的方案是，在批處理任務(wù)內(nèi)部生成全新的數(shù)據(jù)庫，并將其以文件的形式寫入分布式系統(tǒng)的文件夾中。一旦任務(wù)成功執(zhí)行，這些數(shù)據(jù)文件就會(huì)稱為不可變的（immutable），并且可以批量加載（bulk loading）進(jìn)只處理只讀請(qǐng)求的服務(wù)中。很多 KV 存儲(chǔ)都支持使用 MapReduce 任務(wù)構(gòu)建數(shù)據(jù)庫文件，比如 Voldemort，Terrapin， ElephantDB 和 HBase bulk loading。另外 RocksDB 支持 ingest SST 文件，也是類似的情況。

直接構(gòu)建數(shù)據(jù)庫底層文件，就是一個(gè) MapReduce 應(yīng)用的絕佳案例：使用 Mapper 抽取 key，然后利用該 key 進(jìn)行排序，已經(jīng)覆蓋了構(gòu)建索引中的大部分流程。由于大部 KV 存儲(chǔ)都是只讀的（通過批處理任務(wù)一次寫入后，即不可變），這些存儲(chǔ)的底層數(shù)據(jù)結(jié)構(gòu)可以設(shè)計(jì)的非常簡單。例如，不需要 WAL（參見讓 B 樹更可靠）。

當(dāng)數(shù)據(jù)加載進(jìn) Voldemort 時(shí)，服務(wù)器可以利用老文件繼續(xù)對(duì)外提供服務(wù)，新文件會(huì)從分布式文件系統(tǒng)中拷貝的 Voldemort 服務(wù)本地。一旦拷貝完成，服務(wù)器可以立即將外部查詢請(qǐng)求原子地切到新文件上。如果導(dǎo)入過程中發(fā)生了任何問題，也可以快速地切回，使用老文件提供服務(wù)。因?yàn)槔衔募遣豢勺兊模覜]有立即被刪除。

批處理輸出的哲學(xué)

本章稍早我們討論過 Unix 的設(shè)計(jì)哲學(xué)，它鼓勵(lì)在做實(shí)驗(yàn)時(shí)使用顯式的數(shù)據(jù)流：每個(gè)程序都會(huì)讀取輸入，然后將輸出寫到其他地方。在這個(gè)過程中，輸入保持不變，先前的輸出被變換為新的輸出，并且沒有任何其他的副作用。這意味著，你可以任意多次的重新跑一個(gè)命令，每次可以對(duì)命令或者參數(shù)進(jìn)行下微調(diào)，或者查看中間結(jié)果進(jìn)行調(diào)試，而不用擔(dān)心對(duì)你原來系統(tǒng)的狀態(tài)造成任何影響。

MapReduce 任務(wù)在處理輸出時(shí)，遵從同樣的哲學(xué)。通過不改變輸入、不允許副作用（比如輸出到外部文件），批處理不僅可以獲得較好的性能，同時(shí)也變得容易維護(hù)：

• 容忍人為錯(cuò)誤。如果你在代碼中不小心引入了 bug，使得輸出出錯(cuò)，你可以簡單地將代碼回滾到最近一個(gè)正確的版本，然后重新運(yùn)行任務(wù)，則輸出就會(huì)變正確?；蛘?，更簡單地，你可將之前正確的輸出保存在其他的文件夾，然后在遇到問題時(shí)簡單的切回去即可。使用讀寫事務(wù)的數(shù)據(jù)庫是沒法具有這種性質(zhì)的：如果你部署了有 bug 的代碼，并且因此往數(shù)據(jù)庫中寫入了錯(cuò)誤的數(shù)據(jù)，回滾代碼版本也并不能修復(fù)這些損壞的數(shù)據(jù)。（從有 bug 的代碼中恢復(fù)，稱為容忍人為錯(cuò)誤，human fault tolerance）。這其實(shí)是通過犧牲空間換來的，也是經(jīng)典的增量更新而非原地更新。
• 便于敏捷開發(fā)。相比可能會(huì)造成不可逆損壞的環(huán)境，由于能夠很方便地進(jìn)行回滾，可以大大加快功能迭代的速度（因?yàn)椴恍枰M(jìn)行嚴(yán)密的測(cè)試即可上生產(chǎn)）。最小化不可逆性（minimizing irreversibility）的原則，有助于敏捷軟件開發(fā)。
• 簡單重試就可以容錯(cuò)。如果某個(gè) map 或者 reduce 任務(wù)失敗了，MapReduce 框架會(huì)自動(dòng)在相同輸入上對(duì)其重新調(diào)度。如果失敗是由代碼 bug 引起的，在重試多次后（可以設(shè)置某個(gè)閾值），會(huì)最終引起任務(wù)失??；但如果失敗是暫時(shí)的，該錯(cuò)誤就能夠被容忍。這種自動(dòng)重試的機(jī)制之所以安全，是因?yàn)檩斎胧遣豢勺兊?，且失敗子任?wù)的輸出會(huì)被自動(dòng)拋棄。
• 數(shù)據(jù)復(fù)用。同一個(gè)文件集能夠作為不同任務(wù)的輸入，包括用于計(jì)算指標(biāo)的監(jiān)控任務(wù)、評(píng)估任務(wù)的輸出是否滿足預(yù)期性質(zhì)（如，和之前一個(gè)任務(wù)的比較并計(jì)算差異）。
• 邏輯布線分離。和 Unix 工具一樣，MapReduce 也將邏輯和接線分離（通過配置輸入、輸出文件夾），從而分拆復(fù)雜度并且提高代碼復(fù)用度：一些團(tuán)隊(duì)可以專注于實(shí)現(xiàn)干好單件事的任務(wù)開發(fā)；另一些團(tuán)隊(duì)可以決定在哪里、在何時(shí)來組合跑這些代碼。

在上述方面，Unix 中用的很好地一些設(shè)計(jì)原則也適用 Hadoop——但 Unix 工具和 Hadoop 也有一些不同的地方。比如，大部分 Unix 工具假設(shè)輸入輸出是無類型的文本，因此不得不花一些時(shí)間進(jìn)行輸入解析（比如之前的例子中，需要按空格分割，然后取第 7 個(gè)字段，以提取 URL）。在 Hadoop 中，通過使用更結(jié)構(gòu)化的數(shù)據(jù)格式，消除了底層的一些低價(jià)值的語法解析和轉(zhuǎn)換：Avro （參見Avro）和 Parquet 是較常使用的兩種編碼方式，他們提供基于模式的高效編碼方式，并且支持模式版本的演進(jìn)。

對(duì)比 Hadoop 和分布式數(shù)據(jù)庫

從之前討論我們可以感覺到，Hadoop 很像一個(gè)分布式形態(tài)的 Unix。其中，HDFS 對(duì)標(biāo) Unix 中的文件系統(tǒng)，MapReduce 類似于 Unix 進(jìn)程的一個(gè)奇怪實(shí)現(xiàn)（在 map 階段和 reduce 階段間必須要進(jìn)行排序）。在這些源語之上，我們可以實(shí)現(xiàn)各種 join 和 group 語義。

MapReduce 被提出時(shí)，并非是一種全新的思想。在其十多年前，所有前述小節(jié)我們提到的一些并行 join 算法都已經(jīng)被 MPP （massive parallel processing）數(shù)據(jù)庫所實(shí)現(xiàn)了。如，Gamma data base machine、Teradata 和 Tandem NonStop SQL 都是這個(gè)領(lǐng)域的先驅(qū)。

當(dāng)然，如果硬要區(qū)分的話：

1. MPP 數(shù)據(jù)庫是在一組機(jī)器上分布式地、并行執(zhí)行分析型的 SQL
2. MapReduce 和分布式文件系統(tǒng)提供了一種類似于操作系統(tǒng)的、更為通用的計(jì)算方式

存儲(chǔ)類型更為多樣

數(shù)據(jù)庫要求用戶遵循特定的模式（schema，數(shù)據(jù)模型，如關(guān)系型或者文檔型）組織數(shù)據(jù)，但分布式系統(tǒng)中的文件是面向字節(jié)序列（byte arrary，即內(nèi)容對(duì)于系統(tǒng)是黑盒），用戶可以使用任何必要的方式進(jìn)行建模和編碼。因此，這些文件既可以是數(shù)據(jù)庫記錄的集合，也可以是文本、圖像、視頻、傳感器數(shù)值、稀疏矩陣、特征向量、基因序列，或者其他任意類型的數(shù)據(jù)。

換句話說，Hadoop 允許你以任意格式的數(shù)據(jù)灌入 HDFS，將如何處理的靈活性推到之后（對(duì)應(yīng)之前討論過的 schema-less，或者 schema-on-read ）。與之相反，MPP 數(shù)據(jù)庫通常要求用戶在數(shù)據(jù)導(dǎo)入之前，就要針對(duì)數(shù)據(jù)類型和常用查詢模式，進(jìn)行小心的建模（對(duì)應(yīng) schema-on-write）。

從完美主義者的角度來說，事先對(duì)業(yè)務(wù)場景進(jìn)行仔細(xì)地建模再導(dǎo)入數(shù)據(jù)才是正道。只有這樣，數(shù)據(jù)庫用戶才能夠得到更高質(zhì)量的數(shù)據(jù)。然而在實(shí)踐中，即便不管模式快速導(dǎo)入數(shù)據(jù)，可能會(huì)讓數(shù)據(jù)處于奇怪、難用、原始的格式，反而會(huì)比事先規(guī)劃、考究建模后將限制死格式更為有價(jià)值。

這種思想和數(shù)據(jù)庫倉庫很像：在大型組織中，將從不同部門來的數(shù)據(jù)快速聚集到一塊非常重要，因?yàn)檫@提供了將原先分離的數(shù)據(jù)進(jìn)行聯(lián)結(jié)（join）的各種可能性。MPP 數(shù)據(jù)庫所要求的小心精確地建模，會(huì)嚴(yán)重拖慢中心化數(shù)據(jù)的速度。以原始格式將數(shù)據(jù)聚集到一塊，之后再去考慮如何進(jìn)行建模，可以大大加速數(shù)據(jù)收集速度（這個(gè)概念有時(shí)也被稱為數(shù)據(jù)湖，data lake，或者企業(yè)數(shù)據(jù)中心，enterprise data hub）。

無腦數(shù)據(jù)導(dǎo)入其實(shí)是將數(shù)據(jù)理解的復(fù)雜度進(jìn)行了轉(zhuǎn)移：數(shù)據(jù)生產(chǎn)者無需關(guān)心數(shù)據(jù)會(huì)被如何使用，這是數(shù)據(jù)消費(fèi)者的問題（類似讀時(shí)模式，參見文檔模型中 Schema 的靈活性）。在數(shù)據(jù)生產(chǎn)者和消費(fèi)者處于不同團(tuán)隊(duì)、具有不同優(yōu)先級(jí)時(shí)，這種方式的優(yōu)勢(shì)非常明顯。因?yàn)榭赡軟]有一種通用的理想模型，出于不同目的，會(huì)有不同的看待數(shù)據(jù)方式。將數(shù)據(jù)以原始方式導(dǎo)入，允許之后不同消費(fèi)者進(jìn)行不同的數(shù)據(jù)變換。這種方式被總結(jié)為 sushi 原則：數(shù)據(jù)越原始越好。

因此 Hadoop 經(jīng)常用于 ETL 處理：將數(shù)據(jù)以某種原始的格式從事務(wù)型的處理系統(tǒng)中引入到分布式文件系統(tǒng)中，然后編寫 MapReduce 任務(wù)以處理這些數(shù)據(jù)，將其轉(zhuǎn)換回關(guān)系形式，進(jìn)而導(dǎo)入到 MPP 數(shù)據(jù)倉庫匯總以備進(jìn)一步分析之用。數(shù)據(jù)建模依然存在，但是拆解到了其他的步驟，從而與數(shù)據(jù)收集解耦了開來。由于分布式文件系統(tǒng)不關(guān)心應(yīng)用方以何種方式對(duì)數(shù)據(jù)進(jìn)行編碼，僅面向字節(jié)數(shù)組存儲(chǔ)，讓這種解耦成為了可能。

處理模型更為多樣

MPP 數(shù)據(jù)庫是一種將硬盤上的存儲(chǔ)布局、查詢計(jì)劃生成、調(diào)度和執(zhí)行等功能模塊緊密糅合到一塊的整體式軟件。這些組件都可以針對(duì)數(shù)據(jù)庫的特定需求進(jìn)行調(diào)整和優(yōu)化，針對(duì)目標(biāo)查詢類型，系統(tǒng)在整體上可以獲得很好的性能。此外，SQL 作為一種聲明式的查詢語言， 表達(dá)能力很強(qiáng)、語義簡潔優(yōu)雅，讓人可以通過圖形界面而無需編寫代碼就可以完成對(duì)數(shù)據(jù)進(jìn)行訪問。

但從另外的角度來說，并非所有類型的數(shù)據(jù)處理需求都可以合理地表達(dá)為 SQL 查詢。例如，如果你想要構(gòu)建機(jī)器學(xué)習(xí)和推薦系統(tǒng)、支持相關(guān)性排序的全文索引引擎、進(jìn)行圖像分析，則可能需要更為通用的數(shù)據(jù)處理模型。這些類型的數(shù)據(jù)處理構(gòu)建通常都和特定應(yīng)用強(qiáng)耦合（例如，用于機(jī)器學(xué)習(xí)的特征工程、用于機(jī)器翻譯的自然語言模型、用于欺詐預(yù)測(cè)的風(fēng)險(xiǎn)預(yù)估函數(shù)），因此不可避免地需要用通用語言寫代碼來實(shí)現(xiàn)，而不能僅僅寫一些查詢語句。

MapReduce 使工程師能夠在大型數(shù)據(jù)集尺度上輕松的運(yùn)行自己的代碼（而不用關(guān)心底層分布式的細(xì)節(jié)）。如果你已經(jīng)有 HDFS 集群和 MapReduce 計(jì)算框架，你可以基于此構(gòu)建一個(gè) SQL 查詢執(zhí)行引擎， Hive 項(xiàng)目就是這么干的。當(dāng)然，對(duì)于一些不適合表達(dá)為 SQL 查詢的處理需求，也可以基于 Hadoop 平臺(tái)來構(gòu)建一些其他形式的批處理邏輯。

但后來人們又發(fā)現(xiàn)，對(duì)于某些類型的數(shù)據(jù)處理， MapReduce 限制太多、性能不佳，因此基于 Hadoop開發(fā)了各種其他的處理模型（在之后 “MapReduce 之外”小節(jié)中會(huì)提到一些）。僅有 SQL 和 MapReduce 這兩種處理模型是不夠的，我們需要更多的處理模型！由于Hadoop平臺(tái)的開放性，我們可以較為容易實(shí)現(xiàn)各種處理模型。然而，在 MPP 數(shù)據(jù)庫的限制下，我們想支持更多處理模型基本是不可能的。

更為重要的是，基于 Hadoop 實(shí)現(xiàn)的各種處理模型可以共享集群并行運(yùn)行，且不同的處理模型都可以訪問 HDFS 上的相同文件。在 Hadoop 生態(tài)中，無需將數(shù)據(jù)在不同的特化系統(tǒng)間倒來倒去以進(jìn)行不同類型的處理：Hadoop 系統(tǒng)足夠開放，能夠以單一集群支持多種負(fù)載類型。無需移動(dòng)數(shù)據(jù)讓我們更容易的從數(shù)據(jù)中挖掘價(jià)值，也更容易開發(fā)新的處理模型。

Hadoop 生態(tài)系統(tǒng)既包括隨機(jī)訪問型的 OLTP 數(shù)據(jù)庫，如HBase（參見“SSTables和LSM-Trees”），也包括 MPP 風(fēng)格的分析型數(shù)據(jù)庫，例如 Impala。HBase 和 Impala 都不依賴 MapReduce 進(jìn)行計(jì)算，但兩者都使用 HDFS 作為底層存儲(chǔ)。它們?cè)L問數(shù)據(jù)和處理數(shù)據(jù)的方式都非常不同，但卻可以神奇的并存于 Hadoop 生態(tài)中。

面向頻繁出錯(cuò)設(shè)計(jì)

在對(duì)比 MapReduce 和 MPP 數(shù)據(jù)庫時(shí)，我們會(huì)發(fā)現(xiàn)設(shè)計(jì)思路上的兩個(gè)顯著差異：

1. 故障處理方式：取決于對(duì)處理成本、故障頻次的假設(shè)
2. 內(nèi)存磁盤使用：取決于對(duì)數(shù)據(jù)量的假設(shè)

相對(duì)在線系統(tǒng)，批處理系統(tǒng)對(duì)故障的敏感性要低一些。如果批處理任務(wù)失敗，并不會(huì)立即影響用戶，而且可以隨時(shí)重試。

如果在執(zhí)行查詢請(qǐng)求時(shí)節(jié)點(diǎn)崩潰，大多數(shù) MPP 數(shù)據(jù)庫會(huì)中止整個(gè)查詢，并讓用戶進(jìn)行重試或自動(dòng)重試。由于查詢運(yùn)行時(shí)通常會(huì)持續(xù)數(shù)秒或數(shù)分鐘，這種簡單粗暴的重試的處理錯(cuò)誤的方式還可以接受，畢竟成本不算太高。MPP 數(shù)據(jù)庫還傾向?qū)?shù)據(jù)盡可能地存在內(nèi)存里（例如在進(jìn)行 HashJoin 的 HashBuild 時(shí)），以避免讀取磁盤的額外損耗。

與之相對(duì)，MapReduce 在遇到某個(gè) map 或 reduce 子任務(wù)運(yùn)行出錯(cuò)時(shí)，可以單獨(dú)、自動(dòng)地進(jìn)行重試，而不會(huì)引起整個(gè) MapReduce 任務(wù)的重試。此外，MapReduce 傾向于將數(shù)據(jù)（甚至是 map 到 reduce 中間環(huán)節(jié)的數(shù)據(jù)）進(jìn)行落盤，一方面是為了容錯(cuò)，另一方面是因?yàn)?MapReduce 在設(shè)計(jì)時(shí)假設(shè)面對(duì)的數(shù)據(jù)量足夠大，內(nèi)存通常裝不下。

因此，MapReduce 通常更適合大任務(wù)：即那些需要處理大量數(shù)據(jù)、運(yùn)行較長時(shí)間的任務(wù)。而巨量的數(shù)據(jù)、過長的耗時(shí)，都會(huì)使得處理過程中遇到故障司空見慣。在這種情況下，由于一個(gè)子任務(wù)（task） 的故障而重試整個(gè)任務(wù)（job） 就非常得不償失。當(dāng)然，即使只在子任務(wù)粒度進(jìn)行重試，也會(huì)讓那些并不出錯(cuò)的任務(wù)運(yùn)行的更慢（數(shù)據(jù)要持久化）。但對(duì)于頻繁出錯(cuò)的任務(wù)場景來說，這個(gè)取舍是合理的。

但這種假設(shè)在多大程度上是正確的呢？在大多數(shù)集群中，機(jī)器確實(shí)會(huì)故障，但非常低頻——甚至可以低到大多任務(wù)在運(yùn)行時(shí)不會(huì)遇到任何機(jī)器故障。在這種情況下，為了容錯(cuò)引入的巨量額外損耗值得嗎？

為了理解 MapReduce 克制使用內(nèi)存、細(xì)粒度重試的設(shè)計(jì)原因，我們需要回顧下 MapReduce 的誕生歷程。當(dāng)時(shí)谷歌內(nèi)部的數(shù)據(jù)中心很多都是共享使用的——集群中的同一個(gè)機(jī)器上，既有在線的生產(chǎn)服務(wù)，也有離線的批處理任務(wù)。每個(gè)任務(wù)使用容器（虛擬化）的方式進(jìn)行（CPU、RAM、Disk Space）資源預(yù)留。不同任務(wù)之間存在優(yōu)先級(jí)，如果某個(gè)高優(yōu)先級(jí)的任務(wù)需要更多資源，則該任務(wù)所在機(jī)器上的低優(yōu)先級(jí)任務(wù)可能就會(huì)被干掉以讓出資源。當(dāng)然，優(yōu)先級(jí)是和計(jì)算資源的價(jià)格掛鉤的：團(tuán)隊(duì)需要為用到的資源付費(fèi)，高優(yōu)先級(jí)的資源要更貴。

這種架構(gòu)設(shè)計(jì)的好處是，可以面向非線上服務(wù)超發(fā)（overcommitted）資源（這也是云計(jì)算賺錢的理由之一）。因?yàn)橄到y(tǒng)通過優(yōu)先級(jí)跟用戶約定了，在必要時(shí)這些超發(fā)的資源都可以被回收。相比在線離線服務(wù)分開部署，這種混合部署、超發(fā)資源的方式能夠更加充分的利用機(jī)器資源。當(dāng)然代價(jià)就是，以低優(yōu)先級(jí)運(yùn)行的 MapReduce 的任務(wù)可能會(huì)隨時(shí)被搶占。通過這種方式，批處理任務(wù)能夠充分地利用在線任務(wù)等高優(yōu)先級(jí)任務(wù)留下的資源碎片。

統(tǒng)計(jì)來說，在谷歌當(dāng)時(shí)集群中，為了讓位給高優(yōu)先級(jí)任務(wù)，持續(xù)一小時(shí)左右 MapReduce 子任務(wù)大約有 5% 的概率被中止。這個(gè)概率大概比由于硬件問題、機(jī)器重啟和其他原因造成的子任務(wù)重啟要高一個(gè)數(shù)量級(jí)。在這種搶占率下，對(duì)于一個(gè)包含 100 個(gè)子任務(wù)、每個(gè)子任務(wù)持續(xù) 10 分鐘的 MapReduce 任務(wù)來說，在運(yùn)行過程中，有超過一半的概率會(huì)發(fā)生至少一個(gè)子任務(wù)被中止。

這就是為什么 MapReduce 面向頻繁異常中止設(shè)計(jì)的原因：不是為了解決硬件的故障問題，而是給了系統(tǒng)隨意中止子任務(wù)的自由，進(jìn)而在總體上提高計(jì)算集群的資源利用率。

但在開源的集群調(diào)度系統(tǒng)中，可搶占調(diào)度并不普遍。YARN 的 CapacityScheduler 支持搶占以在不同隊(duì)列間進(jìn)行資源的均衡，但到本書寫作時(shí)，YARN、Mesos、Kubernetes 都不支持更為通用的按優(yōu)先級(jí)搶占調(diào)度。在搶占不頻繁的系統(tǒng)中，MapReduce 這種設(shè)計(jì)取舍就不太有價(jià)值了。在下一節(jié)，我們會(huì)考察一些做出不同取舍的 MapReduce 的替代品。

參考資料

[1]DDIA 讀書分享會(huì): https://ddia.qtmuniao.com/