分布式系統在網際網路公司中的應用已經非常普遍，開源軟體層出不窮。hadoop生态系統，從hdfs到hbase，從mapreduce到spark，從storm到spark streaming, heron, flink等等，如何在開源的汪洋中不會迷失自己？本文将從基本概念、架構并結合自己學習工作中的感悟，闡述如何學習分布式系統。由于分布式系統理論體系非常龐大，知識面非常廣博，筆者能力有限，不足之處，歡迎讨論交流。
常見的分布式系統分為資料存儲系統如hdfs，hbase；資料處理計算系統如storm、spark、flink；資料存儲兼分析混合系統，這類系統在資料存儲的基礎上提供了複雜的資料搜尋查詢功能，如elastic search、druid。對于存儲兼計算的系統，我們仍然可以分開分析，是以本文會從資料存儲和計算兩種系統來論述。
文章的大緻結構：第一部分，分布式系統的基本概念；第二、三部分分别詳細論述資料存儲和資料計算系統；最後一部分總結。
概念
-
分布式系統：每個人都在提分布式系統，那麼什麼是分布式系統？其基本概念就是元件分布在網絡計算機上，元件之間僅僅通過消息傳遞來通信并協調行動。
A distributed system is one in which components located at networked computers communicate and coordinate their actions only by passing messages. （摘自分布式系統概念和設計）
- 節點：節點可以了解為上述概念提到的元件，其實完成一組完整邏輯的程式個體，對應于server上的一個獨立程序。一提到節點，就會考慮節點是有狀态還是無狀态的？判斷标準很簡單，該獨立節點是否維護着本地存儲的一些狀态資訊，或者節點是不是可以随時遷移到其他server上而保持節點的行為和以前一緻，如果是的話，則該節點是無狀态，否則是有狀态的。
- 異常：異常處理可以說是分布式系統的核心問題，那麼分布式異常處理相對于單機來說，有什麼不同呢？在單機系統中，對于程式的處理結果是可以預知的，要麼成功，要麼失敗，結果很明确。可在分布式環境中，處理結果除了明确傳回成功或失敗，還有另外一種狀态：逾時，那逾時意味着處理結果完全不确定，有可能成功執行，也有可能執行失敗，也有可能根本沒執行，這給系統開發帶來了很大的難度。其實各種各樣的分布式協定就是保證系統在各種異常情形下仍能正常的工作，是以在學習分布式系統時，要着重看一下文檔異常處理fault-tolerance章節。
-
CAP理論：學習分布式系統中需要重要了解的理論，同時在架構設計中也可以用到這個理論，例如在一些情形下我們可以通過降低一緻性來提高系統的可用性，将資料的每次資料庫更新操作變成批量操作就是典型的例子。
CAP理論，三個字母代表了系統中三個互相沖突的屬性：
- C（Consistency）：強一緻性，保證資料中的資料完全一緻；
- A（Available）：在系統異常時，仍然可以提供服務，注：這兒的可用性，一方面要求系統可以正常的運作傳回結果，另一方面同樣對響應速度有一定的保障；
-
P（Tolerance to the partition of network ）：既然是分布式系統，很多元件都是部署在不同的server中，通過網絡通信協調工作，這就要求在某些節點服發生網絡分區異常，系統仍然可以正常工作。
CAP 理論指出，無法設計一種分布式協定同時完全具備CAP屬性。
從以上CAP的概念我們得出一個結論，在技術選型時，根據你的需求來判斷是需要AP高可用性的系統（容忍傳回不一緻的資料）還是CP強一緻性的系統，或者根據系統提供的參數在AC之間權衡。（可能會有讀者會問，為什麼一定需要P呢？既然是分布式系統，在網絡分區異常情況下仍然正常提供服務是必須的。）
資料存儲系統
當資料量太大以及已經超過單機所能處理的極限時，就需要使用到資料存儲分布式系統。無論是選擇開源系統還是自己設計，第一個要考慮的問題就是資料如何分布式化。
資料分布方式
哈希方式：哈希方式是最常見的資料分布方式。可以簡單想象有一個大的hash表，其中每個桶對應的一台存儲伺服器，每條資料通過某種方式計算出其hash值配置設定到對應的桶中。 int serverId = data.hashcode % serverTotalNum 上面隻是一個簡單的計算公式示例，通過這種方式就可以将資料配置設定到不同的伺服器上。
- 優點：不需要存儲資料和server映射關系的meta資訊，隻需記錄serverId和server ip映射關系即可。
- 缺點：可擴充性不高，當叢集規模需要擴充時，叢集中所有的資料需要遷移，即使在最優情況下——叢集規模成倍擴充，仍然需要遷移叢集一半的資料（這個問題有時間可以考慮一下，為啥隻需要遷移一半？）；另一個問題：資料通過某種hash計算後都落在某台伺服器上，造成資料傾斜（data skew）問題。
- 應用例子：ElasticSearch資料分布就是hash方式，根據routingId取模映射到對應到不同node上。
資料範圍分布：将資料的某個特征值按照值域分為不同區間。比如按時間、區間分割，不同時間範圍劃分到不同server上。
- 優點：資料區間可以自由分割，當出現資料傾斜時，即某一個區間的資料量非常大，則可以将該區間split然後将資料進行重配置設定；叢集友善擴充，當添加新的節點，隻需将資料量多的節點資料遷移到新節點即可。
- 缺點：需要存儲大量的元資訊（資料區間和server的對應關系）。
- 應用例子：Hbase的資料分布則是利用data的rowkey進行區間劃分到不同的region server，而且支援region的split。
資料量分布：按資料量分布，可以考慮一個簡單例子：當使用log檔案記錄一些系統運作的日志資訊時，當日志檔案達到一定大小，就會生成新的檔案開始記錄後續的日志資訊。這樣的存儲方式和資料的特征類型沒有關系，可以了解成将一個大的檔案分成固定大小的多個block。
- 優點：不會有資料傾斜的問題，而且資料遷移時速度非常快（因為一個檔案由多個block組成，block在不同的server上，遷移一個檔案可以多個server并行複制這些block）。
- 缺點： 需要存儲大量的meta資訊（檔案和block的對應關系，block和server的對應關系）。
- 應用例子：Hdfs的檔案存儲按資料量block分布。
一緻性哈希：前文剛提到的哈希方式，當添加删除節點時候，所有節點都會參與到資料的遷移，整個叢集都會受到影響。那麼一緻性哈希可以很好的解決這個問題。一緻性哈希和哈希的資料分布方式大概一緻，唯一不同的是一緻性哈希hash的值域是個環。
- 優點：叢集可擴充性好，當增加删除節點，隻影響相鄰的資料節點。
- 缺點：上面的優點同時也是缺點，當一個節點挂掉時，将壓力全部轉移到相鄰節點，有可能将相鄰節點壓垮。
- 應用例子：Cassandra資料分布使用的是一緻性hash，隻不過使用的是一緻性hash改良版：虛拟節點的一緻性hash（有興趣的可以研究下）。
讨論完資料分布問題，接下來該考慮如何解決當某個節點服務不可達的時候系統仍然可以正常工作（分布式系統CAP中網絡分區異常問題）？這個問題的解決方案說起來很簡單，就是将資料的存儲增加多個副本，而且分布在不同的節點上，當某個節點挂掉的時候，可以從其他資料副本讀取。
引入多個副本後，引來了一系列問題：多個副本之間，讀取時以哪個副本的資料為準呢，更新時什麼才算更新成功，是所有副本都更新成功還是部分副本更新成功即可認為更新成功？這些問題其實就是CAP理論中可用性和一緻性的問題。其中primary-secondary副本控制模型則是解決這類問題行之有效的方法。
primary-secondary控制模型
主從（primary-secondary ）模型是一種常見的副本更新讀取模型，這種模型相對來說簡單，所有的副本相關控制都由中心節點控制，資料的并發修改同樣都由主節點控制，這樣問題就可以簡化成單機問題，極大的簡化系統複雜性。
注：常用的副本更新讀取架構有兩種：主從（primary-secondary）和去中心化（decentralized）結構，其中主從結構較為常見，而去中心化結構常采用paxos、raft、vector time等協定，這裡由于本人能力有限，就不再這兒叙述了，有興趣可以自己學習，歡迎補充。
其中涉及到主從副本操作有以下幾種：
副本的更新
副本更新基本流程：資料更新操作發到primary節點，由primary将資料更新操作同步到其他secondary副本，根據其他副本的同步結果傳回用戶端響應。各類資料存儲分布式系統的副本更新操作流程大體是一樣的，唯一不同的是primary副本更新操作完成後響應用戶端時機的不同，這與系統可用性和一緻性要求密切相關。
以mysql的master slave簡單說明下，通常情況下，mysql的更新隻需要master更新成功即可響應用戶端，slave可以通過binlog慢慢同步，這種情形讀取slave會有一定的延遲，一緻性相對較弱，但是系統的可用性有了保證；另一種slave更新政策，資料的更新操作不僅要求master更新成功，同時要求slave也要更新成功，primary和secondray資料保持同步，系統保證強一緻性，但可用性相對較差，響應時間變長。
上述的例子隻有兩個副本，如果要求強一緻性，所有副本都更新完成才認為更新成功，響應時間相對來說也可以接受，但是如果副本數更多，有沒有什麼方法在保證一定一緻性同時滿足一定的可用性呢？這時就需要考慮Quorum協定，其理論可以用一個簡單的數學問題來說明：
有N個副本，其中在更新時有W個副本更新成功，那我們讀取R個副本，W、R在滿足什麼條件下保證我們讀取的R個副本一定有一個副本是最新資料（假設副本都有一個版本号，版本号大的即為最新資料）？
問題的答案是：W+R > N (有興趣的可以思考下)
通過quorum協定，在保證一定的可用性同時又保證一定的一緻性的情形下，設定副本更新成功數為總副本數的一半（即N/2+1）成本效益最高。（看到這兒有沒有想明白為什麼zookeeper server數最好為基數個？）
副本的讀取
副本的讀取政策和一緻性的選擇有關，如果需要強一緻性，我們可以隻從primary副本讀取，如果需要最終一緻性，可以從secondary副本讀取結果，如果需要讀取最新資料，則按照quorum協定要求，讀取相應的副本數。
副本的切換
當系統中某個副本不可用時，需要從剩餘的副本之中選取一個作為primary副本來保證後續系統的正常執行。這兒涉及到兩個問題：
- 副本狀态的确定以及防止brain split問題：一般方法是利用zookeeper中的sesstion以及臨時節點，其基本原理則是lease協定和定期heartbeat。Lease協定可以簡單了解成參與雙方達成一個承諾，針對zookeeper，這個承諾就是在session有效時間内，我認為你的節點狀态是活的是可用的，如果發生session timeout，認為副本所在的服務已經不可用，無論誤判還是服務真的宕掉了，通過這種機制可以防止腦裂的發生。但這樣會引起另外一個問題：當在session timeout期間，primary 副本服務挂掉了，這樣會造成一段時間内的服務不可用。
- primary副本的确定：這個問題和副本讀取最新資料其實是一個問題，可以利用quoram以及全局版本号确定primary副本。zookeeper在leader選舉的過程中其實利用了quoram以及全局事務id——zxid确定primary副本。
存儲架構模型
關于資料的分布和副本的模型這些細節問題已經詳細叙述，那麼從系統整體架構來看，資料存儲的一般流程和主要子產品都有哪些呢？從中繼資料存儲以及節點之間的membership管理方面來看，主要分以下兩類：
中心化的節點membership管理架構
這類系統主要分為三個子產品：client子產品，負責使用者和系統内部子產品的通信；master節點子產品，負責中繼資料的存儲以及節點健康狀态的管理；data節點子產品，用于資料的存儲和資料查詢傳回。
資料的查詢流程通常分兩步：1. 向master節點查詢資料對應的節點資訊；2. 根據傳回的節點資訊連接配接對應節點，傳回相應的資料。
分析一下目前常見的資料存儲系統，從hdfs，hbase再到Elastic Search，通過與上述通用系統對比，發現：master節點子產品具體對應hdfs的namenode、hbase的hMaster、Elastic Search的master節點；data節點對應hdfs的datanode、hbase的region server、Elastic Search的data node。
去中心化的節點membership管理架構
與上一模型比較，其最大的變化就是該架構中不存在任何master節點，系統中的每個節點可以做類似master的任務：存儲系統元資訊以及管理叢集節點。
資料的查詢方式也有所不同，client可以通路系統中的任意節點，而不再局限于master節點，具體查詢流程如下：1. 查詢系統中任意節點，如果該資料在此節點上則傳回相應的資料，如果不在該節點，則傳回對應資料的節點位址，執行第二步；2. 獲得資料對應的位址後向相關請求資料。
節點之間共享狀态資訊是如何做到的呢？常用的方法是使用如gossip的協定以及在此基礎之上開發的serf架構，感興趣的話可以參考redis cluster 和 consul實作。
資料計算處理系統
常用的資料計算主要分為離線批量計算，可以是實時計算，也可以是準實時mini-batch計算，雖然開源的系統很多，且每個系統都有其側重點，但有些問題卻是共性相通的。
資料投遞政策
在資料進行中首先要考慮一個問題，我們的資料記錄在系統中會被處理幾次（包括正常情形和異常情形）：
- at most once：資料處理最多一次，這種語義在異常情況下會有資料丢失；
- at least once：資料處理最少一次，這種語義會造成資料的重複；
- exactly once：資料隻處理一次，這種語義支援是最複雜的，要想完成這一目标需要在資料處理的各個環節做到保障。
如何做到exactly once， 需要在資料處理各個階段做些保證：
- 資料接收：由不同的資料源保證。
- 資料傳輸：資料傳輸可以保證exactly once。
- 資料輸出：根據資料輸出的類型确定，如果資料的輸出操作對于同樣的資料輸入保證幂等性，這樣就很簡單（比如可以把kafka的offset作為輸出mysql的id），如果不是，要提供額外的分布式事務機制如兩階段送出等等。
異常任務的處理
異常處理相對資料存儲系統來說簡單很多，因為資料計算的節點都是無狀态的，隻要啟動任務副本即可。
注意：異常任務除了那些失敗、逾時的任務，還有一類特殊任務——straggler（拖後腿）任務，一個大的Job會分成多個小task并發執行，發現某一個任務比同類型的其他任務執行要慢很多（忽略資料傾斜導緻執行速度慢的因素）。
其中任務恢複政策有以下幾種：
- 簡單暴力，重新開機任務重新計算相關資料，典型應用：storm，當某個資料執行逾時或失敗，則将該資料從源頭開始在拓撲中重新計算。
- 根據checkpoint重試出錯的任務，典型應用：mapreduce，一個完整的資料處理是分多個階段完成的，每個階段（map 或者reduce）的輸出結果都會儲存到相應的存儲中，隻要重新開機任務重新讀取上一階段的輸出結果即可繼續開始運作，不必從開始重新執行該任務。
背壓——Backpressure
在資料進行中，經常會擔心這樣一個問題：資料處理的上遊消費資料速度太快，會不會壓垮下遊資料輸出端如mysql等。 通常的解決方案：上線前期我們會做詳細的測試，評估資料下遊系統承受的最大壓力，然後對資料上遊進行限流的配置，比如限制每秒最多消費多少資料。其實這是一個常見的問題，現在各個實時資料處理系統都提供了背壓的功能，包括spark streaming、storm等，當下遊的資料處理速度過慢，系統會自動降低上遊資料的消費速度。
對背壓感興趣朋友們，或者有想法自己實作一套資料處理系統，可以參考Reactive Stream，該項目對通用資料處理提供了一種規範，采用這種規範比較有名的是akka。
資料處理通用架構
資料處理的架構大抵是相似的，通常包含以下幾個子產品：
- client： 負責計算任務的送出。
- scheduler ： 計算任務的生成和計算資源的排程，同時還包含計算任務運作狀況的監控和異常任務的重新開機。
- worker：計算任務會分成很多小的task， worker負責這些小task的執行同時向scheduler彙報目前node可用資源及task的執行狀況。
上圖是通用的架構模型圖，有些人會問這是hadoop v1版本的mapreduce計算架構圖，現在都已經yarn模式的新的計算架構圖，誰還用這種模式？哈哈，說的對，但是現在仍然有些處理架構就是這種模型————storm。
不妨把圖上的一些概念和storm的概念映射起來：Job tracker 對應于 nimbus，task tracker 對應于 supervisor，每台supervisor 同樣要配置worker slot，worker對應于storm中的worker。 這樣一對比，是不是就覺得一樣了？
這種架構模型有它的問題，責任不明确，每個子產品幹着多樣工作。例如Job tracker不僅要監控任務的執行狀态，還要負責任務的排程。TaskTracker也同樣，不僅要監控task的狀态、執行，同樣還要監控節點資源的使用。
針對以上問題，基于yarn模式的新的處理架構模型，将任務執行狀态的監控和任務資源的排程分開。原來的Job tracker分為resource manger 負責資源的排程，任務執行的監控則交給每個appMaster來負責，原來的task tracker，變為了node manager，負責資源的監控和task的啟動，而task的執行狀态和異常處理則交給appMaster處理。
同樣的，twitter 根據storm架構方面的一些問題，推出了新的處理架構heron，其解決的問題也是将任務的排程和任務的執行狀态監控責任分離，引入了新的概念Topology Master，類似于這兒的appMaster。
總結
分布式系統涵蓋的内容非常多，本篇文章主要從整體架構以及概念上介紹如何入門，學習過程有一些共性的問題，在這兒總結一下：
- 先分析該系統是資料存儲還是計算系統。
- 如果是資料存儲系統，從資料分布和副本政策開始入手；如果是資料處理問題，從資料投遞政策入手。
- 讀對應系統架構圖，對應着常用的架構模型，每個元件和已有的系統進行類比，想一下這個元件類似于hdfs的namenode等等，最後在腦海裡梳理下資料流的整個流程。
- 在了解了系統的大概，着重看下文檔中fault tolerence章節，看系統如何容錯，或者自己可以預先問些問題，比如如果一個節點挂了、一個任務挂了系統是如何處理這些異常的，帶着問題看文檔。
- 文檔詳細讀了一遍，就可以按照官方文檔寫些hello world的例子了，詳細檢視下系統配置項，随着工作的深入就可以看些系統的細節和關鍵源碼了。