一文讀懂分布式系統
我的 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。
以上是我關于分布式存儲系統內容的一些總結，推薦一些相關的論文 ，有興趣的讀者可以看看：
2.Bigtable: A Distributed Storage System for Structured Data.
3.Dynamo: Amazon’s Highly Available Key-value …
4.Introduction to HBase Schema Design
5.Consistency Tradeoffs in Modern Distributed Database System Design
分布式計算系統
聊完了分布式存儲系統，讓我們來聊聊分布式計算系統 ：) 首先解決一個很多初學分布式計算的同學的疑惑：分布式計算和并行計算是一回事嗎?最初我也有這樣的疑惑，而現在我的理解是這樣的：
- 傳統的并行計算要的是：投入更多機器，數據大小不變，計算速度更快。
- 分布式計算要求：投入更多的機器，能處理更大的數據。
換句話說二者的出發點從一開始就不同，一個強調 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 接口具有兩個優點：1. 高效。 2. 實用簡單。 先說說為什么使用簡單。使用 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 等等。由于本人對這個領域并不是很熟，就不詳細介紹了。
以上是我對分布式計算系統的一些介紹，其實每一個方向深入下去都是一個研究領域，在此推薦一些論文：
1.Scaling Distributed Machine Learning with the Parameter Server
2.Distributed GraphLab: A Framework for Machine Learning
3.Piccolo: Building Fast, Distributed Programs with Partitioned ..
4.Dryad: Distributed Data-parallel Programs from Sequential Building …
分布式管理系統：
(未完待續)
