精華: 學習分散式系統需要哪些知識？

我的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