僅需這一篇，能否吃透「分佈式系統」？

我的 PhD 研究方向是分佈式系統，我老闆也是分佈式系統出身，我們實驗室在這方面的積累還算不錯，所以藉此問題談談自己的看法。首先需要說明的是，分佈式系統是一個複雜且寬泛的研究領域，學習一兩門在線課程，看一兩本書可能都是不能完全覆蓋其所有內容的。介於這篇文章是引導初學者入門，所以我個人覺得為初學者介紹一下當前分佈式系統領域的全貌，也許比直接推薦論文和課程更有幫助。當初學者對這個領域建立起一個大的 Picture 之後，可以根據自己的興趣，有選擇性地深入不同領域進行進一步的學習。

本文主要試圖回答以下兩個問題：

1. 近些年分佈式系統領域都在做些什麼。

2. 為什麼現在投入分佈式系統的學習和研究是值得的。

我會盡可能多地去介紹更 “實用” 的分佈式系統知識。

什麼是實用？例如：

• Paxos 是分佈式系統裡一個重要而且實用的技術。
• Consistent Hash 也是分佈式系統裡一個重要而且實用的技術。
• MapReduce、Spark 等等都是很實用的系統。

什麼不實用？ 例如：

• Paxos 算法的數學證明。（注意此處“不實用” 和 “不重要”的區別）

當然，分佈式系統實在是一個太寬泛的話題，本人才疏學淺，回答也僅僅可能側重於我所關心的領域和方向，很多地方都不能面面俱到。所以在此只能拋磚引玉， 蜻蜓點水，歡迎大家提出寶貴意見，我也會及時對文章進行修改和補充。

分佈式系統近些年都在做些什麼？

分佈式系統是一個古老而寬泛的話題，而近幾年因為 “大數據” 概念的興起，又煥發出了新的青春與活力。除此之外，分佈式系統也是一門理論模型與工程技法並重的學科內容。相較於機器學習這樣的研究方向，學習分佈式系統的同學往往會感覺：“入門容易，深入難”。

的確，學習分佈式系統幾乎不需要太多數學知識（相比於機器學習），這也是為什麼會造成 “入門容易” 的錯覺。然而一旦深入下去，往往需要我們去體會 System 研究的 “簡潔” 與 “美”，正如李沐的回答中說的那樣，系統工作是 “藝術” 而不是 “科學” ，這一點我覺得是系統研究工作最難，同時也是最精華的地方。總之把握一點原則：好的系統研究工作，尤其是分佈式系統研究，一定是儘可能地用最簡單、最直觀的方法去解決實際的問題（看看 MapReduce 就知道了），因為簡單就意味著實用。

總體來說，分佈式系統要做的任務就是把多臺機器有機地組合、連接起來，讓其協同完成一件任務，可以是計算任務，也可以是存儲任務。如果一定要給近些年的分佈式系統研究做一個分類的話，我個人認為大概可以包括三大部分：

1. 分佈式存儲系統

2. 分佈式計算系統

3. 分佈式管理系統

近十年來在這三個方向上，毫無疑問， Google 都是開創者，甚至很多業內人士都說，這十年是外界追隨谷歌技術的十年。我們之前說到，分佈式系統的研究是一門由實際問題驅動的研究，而 Google 則是最先需要面對這些實際問題的公司。下面我們分別看看這三個方面工業界以及學術界這幾年都在做些什麼。

分佈式存儲系統：

分佈式存儲系統是一個非常古老的話題，同時也是分佈式系統裡最難，最複雜，涉及面最廣的問題。 往細了分，分佈式存儲系統大概可以分為四個子方向：

1. 結構化存儲

2. 非結構化存儲

3. 半結構化存儲

4. In-memory 存儲

除了這四個子方向之外，分佈式存儲系統還有一系列的理論、算法、技術作為支撐：例如 Paxos、CAP、ConsistentHash、Timing（時鐘）、2PC、3PC 等等，這些內容我們會在後面提到。現在，我們先來看看上述四個子方向大致都在幹些什麼。

結構化存儲（Structured Storage Systems）的歷史非常古老，典型的場景就是事務處理系統或者關係型數據庫（RDBMS）。傳統的結構化存儲都是從單機做起的，比如大家耳熟能詳的 MySQL。有句話說：MySQL 的成長史就是互聯網的成長史。這一點也不為過。除了 MySQL 之外，PostgreSQL 也是近幾年來勢頭非常強勁的一個 RDBMS。我們發現，傳統的結構化存儲系統強調的是：（1）結構化的數據（例如關係表）；（2）強一致性 （例如，銀行系統、電商系統等場景）；（3）隨機訪問（索引、增刪查改、SQL 語言）。然而，正是由於這些性質和限制，結構化存儲系統的可擴展性通常都不是很好，這在一定程度上限制了結構化存儲在大數據環境下的表現。隨著摩爾定律面臨的瓶頸，傳統的單機關係型數據庫系統面臨著巨大的挑戰。不過真的沒辦法了嗎？在此我們先埋下一個伏筆：）

非結構化存儲 （No-structed Storage Systems）：和結構化存儲不同的是，非結構化存儲強調的是高可擴展性，典型的系統就是分佈式文件系統。分佈式文件系統也是一個古老的研究話題，比如 70 年代的 Xerox Alto、80 年代的 NFS、AFS、90 年代 xFS 等等。然而，這些早期的分佈式文件系統只是起到了網絡磁盤的作用，其最大的問題就是不支持容錯 （Fault Tolerance）和錯誤恢復 （Fault Recovery）。而 Google 在 2003 年 SOSP 上推出的 GFS（Google File System）則是做出了里程碑的一步，其開源實現對應為 HDFS。GFS 的主要思想包括：

（1）用 Master 來管理 Metadata。

（2）文件使用 64MB 的 Chunks 來存儲，並且在不同的 Server 上保存多個副本。

（3）自動容錯，自動錯誤恢復。

Google 設計 GFS 最初的目的是為了存儲海量的日誌文件以及網頁等文本信息，並且對其進行批量處理（例如配合 MapReduce 為文檔建立倒排索引，計算網頁 PageRank 等）。和結構化存儲系統相比，雖然分佈式文件系統的可擴展性、吞吐率都非常好，但是幾乎無法支持隨機訪問（Random Access）操作，通常只能進行文件進行追加（Append）操作。而這樣的限制使得非結構化存儲系統很難面對那些低延時，實時性較強的應用。

半結構化存儲 （Semi-structure Storage Systems）的提出便是為了解決非結構化存儲系統隨機訪問性能差的問題。我們通常會聽到一些流行的名詞，比如 NoSQL、Key-Value Store, 甚至包括對象存儲，例如 Protobuf、Thrift 等等。這些都屬於半結構化存儲研究的領域，其中以 NoSQL 近幾年的發展勢頭尤為強勁。NoSQL 系統既有分佈式文件系統所具有的可擴展性，又有結構化存儲系統的隨機訪問能力（例如隨機 Update、Read 操作），系統在設計時通常選擇簡單鍵值（K-V）進行存儲，拋棄了傳統 RDBMS 裡複雜 SQL 查詢以及 ACID 事務。這樣做可以換取系統最大限度的可擴展性和靈活性。在 NoSQL 裡比較有名系統包括：Google 的 Bigtable、Amazon 的 Dynamo，以及開源界大名鼎鼎的 HBase、Cassandra 等。通常這些 NoSQL 系統底層都是基於比較成熟的存儲引擎，比如 Bigtable 就是基於 LevelDB（Jeff dean 寫的，非常好的 C++ 源碼教程），底層數據結構採用 LSM-Tree，除了 LSM-Tree 之外 B-Tree （B+Tree）也是很成熟的存儲引擎數據結構。

In-memory 存儲：隨著業務的併發越來越高，存儲系統對低延遲的要求也越來越高。 同時由於摩爾定律以及內存的價格不斷下降，基於內存的存儲系統也開始普及。In-memory 存儲顧名思義就是將數據存儲在內存中, 從而獲得讀寫的高性能。比較有名的系統包括 Memcahed ，以及 Redis。 這些基於 K-V 鍵值系統的主要目的是為基於磁盤的存儲系統做 Cache。還有一些偏向於內存計算的系統，比如可以追溯到普林斯頓 Kai Lee 教授早期的研究工作 Distributed Shared Memory ( DSM )，斯坦福的 RamCloud，以及最近比較火的基於 Lineage 技術的 Tachyon（Alluxio）項目（Spark 生態系統子項目）等等。

NewSQL：我們在介紹結構化存儲時說到，單機 RDBMS 系統在可擴展性上面臨著巨大的挑戰，然而 NoSQL 不能很好地支持關係模型。那是不是有一種系統能兼備 RDBMS 的特性（例如：完整的 SQL 支持，ACID 事務支持），又能像 NoSQL 系統那樣具有強大的可擴展能力呢？ 2012 年 Google 在 OSDI 上發表的 Spanner，以及 2013 年在 SIGMOD 發表的 F1，讓業界第一次看到了關係模型和 NoSQL 在超大規模數據中心上融合的可能性。不過由於這些系統都太過於黑科技了，沒有大公司支持應該是做不出來的。比如 Spanner 裡用了原子鐘這樣的黑科技來解決時鐘同步問題，打破光速傳輸的限制。在這裡只能對 Google 表示膜拜。

我們在之前提到，分佈式存儲系統有一系列的理論、算法、技術作為支撐：例如 Paxos、CAP、Consistent Hash、Timing（時鐘）、2PC、3PC 等等。那麼如何掌握好這些技術呢？以我個人的經驗，掌握這些內容一定要理解其對應的上下文。什麼意思呢？就是一定要去思考為什麼在當下環境需要某項技術，如果沒有這個技術用其它技術替代是否可行，而不是一味地陷入大量的細節之中。例如：如何掌握好 Paxos? Paxos 本質上來說是一個三階段提交，更 high level 講是一個分佈式鎖。理解 Paxos 必須一步一步從最簡單的場景出發，比如從最簡單的 Master-backup 出發，發現不行；衍生出多數派讀寫，發現還是不行，再到 Paxos。之後再瞭解其變種，比如 Fast Paxos、Multi-Paxos。同理為什麼需要 Consistent Hash，我們可以先思考如果用簡單 Range Partition 劃分數據有什麼問題。再比如學習 2PC、3PC 這樣的技術時，可以想想他們和 Paxos 有什麼關係，能否替代 Paxos。

以上是我關於分佈式存儲系統內容的一些總結，推薦一些相關的論文 ，有興趣的讀者可以看看：

• http://www.eecg.toronto.edu/~ashvin/courses/ece1746/2003/reading/ghemawat-sosp03.pdf
• http://lintool.github.io/UMD-courses/bigdata-2015-Spring/content/ChangFay_etal_OSDI2006.pdf
• https://www.allthingsdistributed.com/files/amazon-dynamo-sosp2007.pdf
• http://lintool.github.io/UMD-courses/bigdata-2015-Spring/content/Khurana_etal_2012.pdf
• http://lintool.github.io/UMD-courses/bigdata-2015-Spring/content/Abadi_2012.pdf
• https://www.usenix.org/conference/osdi12/technical-sessions/presentation/corbett
• https://www.cs.cornell.edu/projects/ladis2009/papers/lakshman-ladis2009.pdf
• https://homes.cs.washington.edu/~billhowe/mapreduce_a_major_step_backwards.html
• http://lintool.github.io/UMD-courses/bigdata-2015-Spring/content/Stonebraker_etal_CACM2010.pdf
• http://www.cs.cmu.edu/~pavlo/courses/fall2013/static/slides/mapreduce.pdf

分佈式計算系統

聊完了分佈式存儲系統，讓我們來聊聊分佈式計算系統 ：） 首先解決一個很多初學分佈式計算的同學的疑惑：分佈式計算和並行計算是一回事嗎？最初我也有這樣的疑惑，而現在我的理解是這樣的：

• 傳統的並行計算要的是：投入更多機器，數據大小不變，計算速度更快。
• 分佈式計算要求：投入更多的機器，能處理更大的數據。

換句話說二者的出發點從一開始就不同，一個強調 High Performance, 一個強調 Scalability。舉例來說，MapReduce 給業界帶來的真正思考是什麼？其實是給我們普及了 Google 這樣級別的公司對真正意義上的「大數據」的理解。因為在 04 年論文出來之前，搞並行計算的人壓根連 「容錯」的概念都沒有。換句話說，分佈式計算最為核心的部分就是「容錯」，沒有容錯，分佈式計算根本無從談起。MapReduce 統要做成這個樣子（Map ＋ Reduce），其實就是為了容錯。

然而很多初學分佈式計算的同學對容錯的概念多多少少是有誤解的。包括我在初學 MapReduce 的時候也會思考：好好的計算怎麼就會出錯了呢？一方面，由於硬件的老化，有可能會導致某臺存儲設備沒有啟動起來，某臺機器的網卡壞了，甚至於計算運行過程中斷電了，這些都是有可能的。然而最頻繁發生的錯誤是計算進程被殺掉。因為 Google 的運行環境是共有集群，任何一個權限更高的進程都可能 Kill 掉你的計算進程。設想在一個擁有幾千臺機器的集群中運行，一個進程都不被 Kill 掉的概率幾乎為零。具體的容錯機制我們會在後面介紹具體的系統時提到。

另一個有意思的話題是，隨著機器學習技術的興起，越來越多的分佈式計算系統是為了機器學習這樣的應用設計的，這也是我比較關注的研究領域，也會在後面重點談到。

如同分佈式存儲系統一樣，我對分佈式計算系統也做了一個分類，如下：

1. 傳統基於 MSG 的系統

2. MapReduce-like 系統

3. 圖計算系統

4. 基於狀態（State）的系統

5. Streaming 系統

當然不同的人可能會有不同的分類方法，不過大同小異。我們接下來聊聊這些系統都在幹些什麼。

傳統基於MSG的系統：這類系統裡比較有代表性的就是 MPI （Message Passing Interface）。目前比較流行的兩個 MPI 實現是 MPICH2 和 OpenMPI。MPI 這個框架非常靈活，對程序的結構幾乎沒有太多約束，以至於大家有時把 MPI 稱為一組接口 API, 而不是系統框架。在這些 API 裡最常用的兩個就是 send 和 recv 接口（還有一系列非阻塞擴展接口，例如：Isend、Irecv 等）。MPI 除了提供消息傳遞接口之外，其框架還實現了資源管理和分配，以及調度的功能。除此之外，MPI 在高性能計算裡也被廣泛使用，通常可以和 Infiniband 這樣的高速網絡無縫結合。

除了 send 和 recv 接口之外，MPI 中另一個接口也值得注意，那就是 AllReduce。這個接口在很多機器學習系統開發裡都很用。因為很多並行機器學習系統都是各個進程分別訓練模型，然後在合適的時候（例如一輪迭代結束）大家同步一下答案，達成共識，然後繼續迭代。這個 “達成共識” 的操作往往可以很方便地通過 AllReduce 來完成。 AllReduce 接口具有兩個優點：高效和使用簡單。 先說說為什麼使用簡單：使用 AllReduce 通常只需要在單機核心源碼里加入 AllReduce 一行代碼，就能完成並行化的功能。說 AllReduce 高效的原因是因為其底層消息傳遞使用了 Tree Aggregation，儘可能地將計算分攤到每一個節點。

可是，既然 AllReduce 這麼好，為什麼在實際大規模計算中很少看到呢？原因很簡單，就是因為 MPI 不支持容錯，所以很難擴展到大規模集群之上。不過最近陳天奇寫了一個支持容錯的 AllReduce 接口，叫 Rabit，有興趣的同學可以關注一下。 大名鼎鼎的 XGBoost 底層的分佈式接口就是 Rabit。

MapReduce-like 系統：這一類系統又叫作 Dataflow 系統，其中以 MapReduce（Hadoop）和 Spark 為代表。其實在學術界有很多類似的系統例如 Dryad、FlumeJava、Twister 等等。這一類系統的特點是將計算抽象成為 High-Level Operator，例如像 Map、Reduce、Filter 這樣的函數式算子，然後將算子組合成 DAG ，然後由後端的調度引擎進行並行化調度。其中，MapReduce 系統屬於比較簡單的 DAG，只有 Map 和 Reduce 兩層節點。MapReduce 這樣的系統之所以可以擴展到超大規模的集群上運行，就是因為其完備的容錯機制。在 Hadoop 社區還有很多基於 MapReduce 框架的衍生產品，比如 Hive（並行數據庫 OLAP）、Pig（交互式數據操作）等等。

MapReduce-like 的編程風格和 MPI 截然相反。MapReduce對程序的結構有嚴格的約束——計算過程必須能在兩個函數中描述：Map 和 Reduce；輸入和輸出數據都必須是一個一個的 Records；任務之間不能通信，整個計算過程中唯一的通信機會是 Map Phase 和 Reduce Phase 之間的 Shuffuling Phase，這是在框架控制下的，而不是應用代碼控制的。因為有了嚴格的控制，系統框架在任何時候出錯都可以從上一個狀態恢復。Spark 的 RDD 則是利用 Lineage，可以讓數據在內存中完成轉換。

由於良好的擴展性，許多人都將機器學習算法的並行化任務放在了這些平臺之上。比較有名的庫包括 Mahout（基於 Hadoop），以及 MLI （基於 Spark) 。然而這些系統最大缺點有兩點：

1. 這些系統所能支持的機器學習模型通常都不是很大。導致這個問題的主要原因是這系統在 push back 機器學習模型時都是粗粒度地把整個模型進行回傳，導致了網絡通信的瓶頸。有些機器學習的模型可以大到無法想象，比如我們用 Field-aware Factorization Machine （FFM）做 Criteo 的 CTR Prediction 時模型大小可以達到 100 GB.

2. 嚴格的 BSP 同步計算使得集群的效率變得很低。也就是說系統很容易受到 straggle 的影響。

圖計算系統：圖計算系統是分佈式計算裡另一個分支，這些系統都是把計算過程抽象成圖，然後在不同節點分佈式執行，例如 PageRank 這樣的任務，很適合用圖計算系統來表示。最早成名的圖計算系統當屬 Google 的 Pregel，該系統採用 BSP 模型，計算以 Vectex 為中心。隨後又有一系列圖計算框架推出，例如：GPS （對 Pregel 做了優化，除了 Vectex-centric Computation，還有 Global Computation，動態調整分區等等。）Giraph / Hama 都是基於 Hadoop 的 Apache 的開源 BSP 圖計算項目。

除了同步（BSP）圖計算系統之外，異步圖計算系統裡的佼佼者當屬 GraphLab，該系統提出了 GAS 的編程模型。目前這個項目已經改名為 Dato，專門推廣基於圖的大規模機器學習系統。

基於狀態（State）的系統：這一類系統主要包括 2010 年 OSDI 上推出的 Piccolo，以及後來 2012 年 NIPS 上 Google 推出的 DistBelief，再到後來被機器系學習領域廣泛應用的 Parameter Server 架構。這裡我們重點介紹一下 Parameter Server 這個架構。

我們之前說，MPI 由於不支持容錯所以很難擴展至大規模集群之中；MapReduce 系統無法支持大模型機器學習應用，並且節點同步效率較低。用圖抽象來做機器學習任務，很多問題都不能很好地求解，比如深度學習中的多層結構。而 Parameter Server 這種 State-Centric 模型則把機器學習的模型存儲參數上升為主要組件，並且採用異步機制提升處理能力。參數服務器的概念最早來自於 Alex Smola 於 2010 年提出的並行 LDA 架構。它通過採用分佈式的 Memcached 作為存放參數的存儲，這樣就提供了有效的機制作用於不同Worker節點同步模型參數。Google 的 Jeff Dean 在 2012 年進一步提出了第一代 Google Brain 大規模神經網絡的解決方案 Distbelief。後來 CMU 的 Eric xing 以及百度少帥李沐都提出了更通用的 Parameter server 架構。

如果要深入 Parameter Server 系統的設計，需要一些機器學習的背景，比如什麼是 SSP 協議， 在此我們就不詳細討論了。

Streaming 系統：Streaming 系統聽名字就能看出來是為流式數據提供服務的。其中比較有名的系統包括 Storm、Spark Streaming、Flink 等等。由於本人對這個領域並不是很熟，就不詳細介紹了。

以上是我對分佈式計算系統的一些介紹，其實每一個方向深入下去都是一個研究領域，在此推薦一些論文：

• MapReduce: Simplified Data Processing on Large Clusters
• Resilient Distributed Datasets
• Scaling Distributed Machine Learning with the Parameter Server
• Distributed GraphLab: A Framework for Machine Learning
• Piccolo: Building Fast, Distributed Programs with Partitioned ..
• Petuum: A New Platform for Distributed Machine Learning on Big Data
• Spark Streaming
• Dryad: Distributed Data-parallel Programs from Sequential Building ...
• Large Scale Distributed Deep Networks - NIPS Proceedings

閱讀更多 互聯網弟弟 的文章

關鍵字: MapReduce 吃透 數據庫