Author：放翁（文初）

Email： fangweng@taobao.com

Mblog:weibo.com/fangweng

Blog: http://blog.csdn.net/cenwenchu79/

Beatles: https://github.com/cenwenchu/beatles

讀前先看：

這篇文章主要講述的是beatles流式數據分析框架中對于master的橫向擴展真實的設計演進，是beatles框架介紹的第三篇，看第三部分的時候如果看過前兩篇文章，這篇文章的遞進應該比較容易理解（如果看過前面的代碼，那么會更深入的理解其中的細節，文字圖片描述一分鐘，代碼寫寫1個周）。如果沒有看過前兩篇，那前提你要理解hadoop等常用的分布式分析系統再看，否則最后可能交流起來就有些空對空了，因為真的寫了和用了就會有體會細節的差別。

廢話不多說，看完后如果不看代碼其實也很難體會細節，因此建議看完后看看代碼，跑一下測試用例子（MasterSlaveIntegrationTest_SocketVersion是socket集成測試版本）。

下面文章中的mr表示MapReduce的意思。

Master的橫向擴展：

盡管Beatles這種松散模式下Slave可以隨時隨地的加入集群，但由于最終數據還要匯總到Master，Master本身承擔著類似于Hadoop中Reduce的角色，所以Master在版本迭代的過程中不斷的對本身做著各種優化：并行模式下多線程的數據合并，主干數據分析周期的磁盤換入換出，支持Slave多任務合并后返回，數據導出廣度遍歷column存儲結構等等（詳細的參看第二篇文章）。當數據base真的非常大的時候（例如：對于用戶緯度的統計，統計后結果還是非常多，無法hold在內存），開放平臺分析系統給數據需求方提供的解決方案是片段性輸出，交由外部再次合并這些片段。（這是基于當時間片段足夠小的時候，數據片內容可以被承受）但結果是讓外部數據使用者利用數據庫去對這些結果做二次歸并。這不僅給數據結果使用者帶來了問題，也使得Master隨時會被最后一根稻草壓倒（如果時間片數據依然無法被hold）。

下面看看老的結構圖：

圖1Master橫向擴展前結構圖

1.master從不同的jobs來源構建需要處理的分析任務，并拆分為多個task是。

2.slave請求任務（一個或者多個）去做分析處理。

3.slave獲得任務后從任務描述中獲得數據來源，分析規則，輸出，開始從數據來源增量獲取數據進行分析。

4.slave根據任務定義判斷多個任務結果是否可以合并，并將合并后的結果發回給master。至此slave的一輪業務生命周期結束，再從步驟3開始重復。

5.master收到各個slave的數據，開始多線程并行合并job結果，最終判斷某一個job的task是否都已經完成，如果完成開始導出數據和臨時文件，重置job開始下一輪的job執行。

思考：

是否增加一個角色：reducer來替代掉master的工作？

其實slave一次能夠獲取多個jobtask，然后自我合并，就是一種比較弱的reduce的做法。

否定了新增一個reducer的原因：新增角色增加了管理的復雜度和集群擴展性。（Hadoop就直接用slave作為reduce）

如果用slave來完全承擔所有的reduce工作？

1.破壞了原來master不管理slave集群的基本原則，slave動態擴展非常麻煩，同時需要增加心跳管理和各種調度算法，任務管理來完成結果的合并。（最后就是一個hadoop的設計）

2.不考慮用文件作為數據交互的方式（因為流式分析的特點：片段性數據量不大，及時性要求高，所以最好直接是內存），因此hadoop最亮眼的hdfs沒有用到，hadoop設計將會大打折扣。

如果用master繼續做reduce，是否可以考慮橫向擴展master？

Master的職責：

1.從任務源（可能是本地配置文件或者db或者是http服務）獲得jobs定義構建任務池。

2.被動分配多個job的tasks。

3.管理job狀態以及jobtask的狀態。

4.根據jobtask狀態合并slave返回結果到job各個主干上。

5.根據job狀態導出job每一輪的中間結果和統計結果。

6.根據job狀態判斷是否重置job。

會發現其實master歸根到底就是對job的管理以及對job中數據結果的合并和導出，而最為消耗的就是類似與reducer的4,5兩個步驟。

先介紹一下job和jobtask的概念，利于對后面的拆分有更好的理解：

job包含了一組jobtask，job本身定義了一組mr規則（類似有很多mr處理實現），定義了要處理的數據來源（其他信息參看代碼）。

Jobtask是將job的多個數據來源拆分后得到的一個子任務，也就是每一個數據來源和同一組mr成為了一個任務，在slave獲得一個或者多個task的時候，可以自己通過數據來源去獲取數據，然后根據一組規則直接對流式數據做大量的mr（這也是和最早hadoop自己寫mr的差別，其實數據的多次移動和讀入才是計算集群的最大成本），但最終所有的jobtask都要合并到job的結果主干上，最后導出臨時結果和報表數據。

a.如果等價部署多個master，所有slave連接一組master，來做master橫向擴展會如何？

a)任務管理就需要多節點之間做并發控制，當前可以看看master內的代碼是一個進程內做并發控制。這種方式過于復雜，帶來的消耗遠大于收益

b.如果等價部署多個master，所有slave連接一組master，但是將job或者jobtask按照業務（上面說過job就是定義了多個mr的實現，要拆分也只能將mr分組放到不同的master上來減輕mr產生的<k,v>對存儲帶來的壓力）分攤到多個master上，即不用對任務管理做并發控制，又可以分擔reduce的工作。

a)slave主動請求任務，如何判斷應該優先向誰請求任務？任務負載均衡如何做？最后可能還是單獨部署多套集群來的靠譜。

b)將不同的mr放在多個master可行，但結果就和hadoop獨立的寫mr帶來的結果一樣，對同一份數據來源處理，卻因為mr的分組數據被反復讀入和移動。

c.和Zookeeper類似，master建立group，但是只有一臺負責1,2,3,6的工作，而4,5則可以擴展到多個master上。需要對原來系統的改造為：

a)多個master job池構建來源保持一致，構建完畢增加mr與master的對應關系（根據算法實現平均分配，后面介紹關于分配的算法，注意只有主的那臺master接受任務分發請求，負責將mr與master建立對應關系在task中傳遞給slave）

b)slave從一臺master上獲取任務，分析后將結果按照mr分組（執行的Task中帶有，所有設計都是這樣，slave不保留任何帶有業務性或者集群管理性的配置，保證slave隨時離場，隨時加入），將分組后的mr結果并行的發送到多個master上。

c)master在合并和輸出結果的時候判斷自己所負責的mr部分，按需合并和存儲（雖然在slave中已經有做業務數據分組）。

會發現多個master好像多臺流水線一樣，保持著相同的動作和周期性，從同一個slave獲取到了不同原始材料以后，開始制作零件，然后以匹配的速度來將不同零件丟到一個筐子里交給后續處理者。

當然你會考慮到還有容錯，master集群動態擴展，速度不匹配等問題，后續細節慢慢介紹。

基于上面所描述的情況，結構演變成如下：

圖2 橫向擴展后的結構圖

可以看到Slave現在獲取任務還是集中在一臺，但是返回任務結果會支持分散到多臺master，解決master瓶頸最大的問題。同時master組的jobs來源保持一致，作為橫向擴展的基礎（主master分組mr簡化master的協同，其他master沒有獲得數據就沒有結果輸出）

細節：

1.master group之間不需要通信。（主要是業務分拆信息，可以通過冪等算法，也可以通過單機分配，分析結果過濾投遞的方式），當前主要是用分析結果過濾投遞來保證。

2.平均分配算法。

首先master有權重（有實體機器，也有虛擬機，處理能力不同），其次mr的權重也不同（有對app做簡單統計的，有對用戶做統計的，數據量相差非常大）。當前考慮一個mr一臺虛擬機抗不論多少數據都能抗的住，如果扛不住后續可以再根據mr產生的結果分（這會增加分流數據的消耗），其實可以看作現在是分庫，以后就是分表。分配算法其實就是在兩邊都有權重的情況下做任務分配，且任務不可切割。

采用了兩個排序隊列，然后按照簡單的饑餓＋權重方式來分配，處理者根據當前獲得的饑餓感排序，越饑餓的放在越前面（饑餓感＝已分配任務／自身權重），當已分配任務為0的時候饑餓感＝1／自身權重（保證能力最強的優先獲得最大的任務）。任務按照權重排序，高權重的排在最前面。分配過程如下：

1.獲的當前任務隊列最前面的任務（權重最高的任務）

2.將任務分配給處理者隊列中饑餓感最強的節點。

3.節點吃了任務以后會再次按照處理者隊列排序。

4.循環到1再次分配任務，直到任務分配結束。

能看到的就是其實就是有一個評判餓的標準，按照資源權重高低來分配，最后平均分配資源。（也許各位會有更好的建議和算法）

這個算法在保證兩個隊列構建時始終一致性的情況下，能夠變成等冪分配算法，其實也就是當兩個隊列中如果遇到評判標準相等的時候排序是否會有前后變化，如果沒有，那么任何一個master都會產出相同的分配結果。（具體可以參看代碼，在ReportUtil中）

3.Master的容災。由于master其實不是等價集群的模式，因此master的不可靠會導致數據丟失。

a)Slave如果發送錯誤，則會嘗試再發送一次，如果兩次錯誤，則將master和對應的job作為文件名保存這次隸屬于這個job的tasks結果到文件中（append 進文件）。

b)Master如果恢復的話，可以通過腳本將這些文件復制到master的目錄下，master后臺線程載入數據合并到主干。

c)過程中如果master恢復，后續的消息將會投遞到master，因此不會再寫這個文件，因此文件不會需要有一個增量拷貝的過程，同時master也可以在后臺線程慢慢恢復合并，最后使得數據保持一致性。

d)當前還是處于半自動模式，后續可以考慮讓slave后臺線程將數據發送到恢復了的master上。

e)中間可能損失部分時間片數據。

4.mr的動態增加或者減少（隨時隨地可以針對流式數據產出新的統計分析結果）。原來的集群就支持mr的動態增加，因為都是配置文件修改，重新載入翻譯一下即可（統計模型被抽象后mr就可以窮舉了，具體參看前面的文檔）。當前因為master是多個，同時開始的時候就做好了mr與master的對應關系，因此增加或者減少如果從新做mr與master的分配會產生數據遷移的需求，因此對于mr的增加只是將變化部分再對master group做一次分配（假設原來那些mr分配是均勻的，現在歸零master再分配，大志結果也是均勻的），對于mr的減少，只是消除掉task中的定義，mr與master對應關系不消除，避免后面要恢復帶來的數據遷移。

5.master的動態增加和減少。這個不可避免的要做數據遷移，當前做法是每天是所有job重置的周期，增加和減少master將在那個時候對整個集群做停機，然后啟動集群做重新編譯，從今天起點開始分析，追趕數據。以后可以考慮如何做到不停機擴容master。（主要就是數據遷移）

6.后續考慮做master與mr的冗余分配，既可以保證數據可靠性（多份數據分析存儲），又可以方便擴容（數據遷移成本可以間接降低）

一些感受

最后master的簡單橫向擴展模式，使得數據分片，一定程度上得到了數據安全性的保證，同時對于非常消耗cpu和memory的reduce被簡單的拆分開來，為業務發展提供了突破，最重要的是系統依然保持最初的設計原則和目的。

任何系統都有自己的適用場景，不要因為要做一個大而全的系統，而喪失了自己設計的原則，我們很難做一個hadoop，但是我們要的也并不是一個通用的hadoop，找到業務場景的特點，才能夠用最簡單高效的設計方式來完成業務不斷增長帶來的技術挑戰。

另，在設計過程中時不時有同學問我要不要引入zookeeper，過程中曾考慮過，但最后還是覺得要解決了根本問題以后再引入，因為它只是一個工具，便于管理的工具。就像我們要求代碼能夠做單元測試一樣，如果沒有zookeeper是否你的系統就無法run，小到模塊，大到系統，接口化設計就是要屏蔽實現對系統設計后期可擴展，穩定性的影響，所有系統最后都能夠退化為文件交換方式的處理模式，因此如果能夠用文件交換可以實現的了的話，你的系統就是最ok的了。（你可以發現linux這么偉大的操作系統就是如此，一切文件化，一切接口化，簡單就是美，這個美需要不斷的堅持自己的原則）

Beatles小記（三）－分布式數據流分析中Master的橫向擴展