從Storm和Spark 學(xué)習(xí)流式實(shí)時分布式計(jì)算的設(shè)計(jì)

【編者按】流式實(shí)時分布式計(jì)算系統(tǒng)在互聯(lián)網(wǎng)公司占有舉足輕重的地位，尤其在在線和近線的海量數(shù)據(jù)處理上。而處理這些海量數(shù)據(jù)的，就是實(shí)時流式計(jì)算系統(tǒng)。Spark是實(shí)時計(jì)算的系統(tǒng)，支持流式計(jì)算，批處理和實(shí)時查詢。除了Spark，流式計(jì)算系統(tǒng)最有名的就是Twitter的Storm和Yahoo的S4。作者參考Storm和Spark探討流式計(jì)算系統(tǒng)的設(shè)計(jì)要點(diǎn)。本文來自CSDN博客。

---

免費(fèi)訂閱“CSDN大數(shù)據(jù)”微信公眾號，實(shí)時了解最新的大數(shù)據(jù)進(jìn)展！

CSDN大數(shù)據(jù)，專注大數(shù)據(jù)資訊、技術(shù)和經(jīng)驗(yàn)的分享和討論，提供Hadoop、Spark、Imapala、Storm、HBase、MongoDB、Solr、機(jī)器學(xué)習(xí)、智能算法等相關(guān)大數(shù)據(jù)觀點(diǎn)，大數(shù)據(jù)技術(shù)，大數(shù)據(jù)平臺，大數(shù)據(jù)實(shí)踐，大數(shù)據(jù)產(chǎn)業(yè)資訊等服務(wù)。

---

以下為原文：

背景

最近我在做流式實(shí)時分布式計(jì)算系統(tǒng)的架構(gòu)設(shè)計(jì)，而正好又要參見CSDN博文大賽的決賽。本來想就寫Spark源碼分析的文章吧。但是又想畢竟是決賽，要拿出一些自己的干貨出來，僅僅是源碼分析貌似分量不夠。因此，我將最近一直在做的系統(tǒng)架構(gòu)的思路整理出來，形成此文。為什么要參考Storm和Spark，因?yàn)闆]有參照效果可能不會太好，尤其是對于Storm和Spark由了解的同學(xué)來說，可能通過對比，更能體會到每個具體實(shí)現(xiàn)背后的意義。

本文對流式系統(tǒng)出現(xiàn)的背景，特點(diǎn)，數(shù)據(jù)HA，服務(wù)HA，節(jié)點(diǎn)間和計(jì)算邏輯間的消息傳遞，存儲模型，計(jì)算模型，與生產(chǎn)環(huán)境融合都有涉及。希望對大家的工作和學(xué)習(xí)有所幫助。

正文開始：

流式實(shí)時分布式計(jì)算系統(tǒng)在互聯(lián)網(wǎng)公司占有舉足輕重的地位，尤其在在線和近線的海量數(shù)據(jù)處理上。在線系統(tǒng)負(fù)責(zé)處理在線請求，因此低延時高可靠是核心指標(biāo)。在線系統(tǒng)是互聯(lián)網(wǎng)公司的核心，系統(tǒng)的好壞直接影響了流量，而流量對互聯(lián)網(wǎng)公司來說意味著一切。在線系統(tǒng)使用的數(shù)據(jù)是來自于后臺的計(jì)算系統(tǒng)產(chǎn)生的。

對于在線（區(qū)別于響應(yīng)互聯(lián)網(wǎng)用戶請求的在線系統(tǒng)，這個在線系統(tǒng)主要是內(nèi)部使用的，也就是說并不直接服務(wù)于互聯(lián)網(wǎng)用戶）/近線系統(tǒng)來說，處理的是線上產(chǎn)生的數(shù)據(jù)，比如在線系統(tǒng)產(chǎn)生的日志，記錄用戶行為的數(shù)據(jù)庫等，因此近線系統(tǒng)也需要低延時高可靠的處理海量數(shù)據(jù)。對于那些時效性很強(qiáng)的數(shù)據(jù)，比如新聞熱點(diǎn)，電商的促銷，微博熱詞等都需要在很短的時間內(nèi)完成數(shù)據(jù)處理以供在線系統(tǒng)使用。

而處理這些海量數(shù)據(jù)的，就是實(shí)時流式計(jì)算系統(tǒng)。Spark是實(shí)時計(jì)算的系統(tǒng)，支持流式計(jì)算，批處理和實(shí)時查詢。它使用一個通用的stack解決了很多問題，畢竟任何公司都想要Unified的平臺去處理遇到的問題，可以減少開發(fā)和維護(hù)的人力成本和部署平臺的物力成本。除了Spark，流式計(jì)算系統(tǒng)最有名的就是Twitter的Storm和Yahoo的S4（其實(shí)Spark的流式計(jì)算還是要弱于Storm的，個人認(rèn)為互聯(lián)網(wǎng)公司對于Storm的部署還是多于Spark的）。

本文主要探討流式計(jì)算系統(tǒng)的設(shè)計(jì)要點(diǎn)，并且通過對Spark和Storm的實(shí)現(xiàn)來給出實(shí)例。通過對于系統(tǒng)設(shè)計(jì)要點(diǎn)的梳理，也可以幫助我們更好的學(xué)習(xí)這些系統(tǒng)的實(shí)現(xiàn)。最后，看一下國內(nèi)互聯(lián)網(wǎng)公司對于這些流式系統(tǒng)的應(yīng)用（僅限于公開發(fā)表的內(nèi)容）。

流式計(jì)算的背景和特點(diǎn)

現(xiàn)在很多公司每天都會產(chǎn)生數(shù)以TB級的大數(shù)據(jù)，如何對這些數(shù)據(jù)進(jìn)行挖掘，分析成了很重要的課題。比如：

1. 電子商務(wù)：需要處理并且挖掘用戶行為產(chǎn)生的數(shù)據(jù)，產(chǎn)生推薦，從而帶來更多的流量和收益。最理想的推薦就是根據(jù)興趣推薦給用戶本來不需要的東西！而每天處理海量的用戶數(shù)據(jù)，需要一個低延時高可靠的實(shí)時流式分布式計(jì)算系統(tǒng)。
2. 新聞聚合：新聞時效性非常重要，如果在一個重大事情發(fā)生后能夠?qū)崟r的推薦給用戶，那么肯定能增大用戶粘性，帶來可觀的流量。
3. 社交網(wǎng)站：大家每天都會去社交網(wǎng)站是為了看看現(xiàn)在發(fā)生了什么，周圍人在做什么。流式計(jì)算可以把用戶關(guān)注的熱點(diǎn)聚合，實(shí)時反饋給用戶，從而達(dá)到一個圈子的聚合效果。
4. 交通監(jiān)管部門：每個城市的交通監(jiān)管部門每天都要產(chǎn)生海量的視頻數(shù)據(jù)，這些視頻數(shù)據(jù)也是以流的形式源源不斷的輸系統(tǒng)中。實(shí)時流式計(jì)算系統(tǒng)需要以最快的速度來處理這些數(shù)據(jù)。
5. 數(shù)據(jù)挖掘和機(jī)器學(xué)習(xí)：它們實(shí)際上是互聯(lián)網(wǎng)公司內(nèi)部使用的系統(tǒng)，主要為線上服務(wù)提供數(shù)據(jù)支撐。它們可以說是互聯(lián)網(wǎng)公司的最核心的平臺之一。系統(tǒng)的效率是挖掘的關(guān)鍵，理想條件下就是每天產(chǎn)生的海量數(shù)據(jù)都能得到有效處理，對于原來的數(shù)據(jù)進(jìn)行全量更新。
6. 大型集群的監(jiān)控：自動化運(yùn)維很重要，集群監(jiān)控的實(shí)時預(yù)警機(jī)制也非常重要，而流式系統(tǒng)對于日志的實(shí)時處理，往往是監(jiān)控系統(tǒng)的關(guān)鍵。
7. 等等。

流式實(shí)時分布式計(jì)算系統(tǒng)就是要解決上述問題的。這些系統(tǒng)的共同特征是什么？

1. 非常方便的運(yùn)行用戶編寫的計(jì)算邏輯：就如Hadoop定義了Map和Reduce的原語一樣，這些系統(tǒng)也需要讓用戶關(guān)注與數(shù)據(jù)處理的具體邏輯上，他們不應(yīng)該也不需要去了解這些usder defined codes是如何在分布式系統(tǒng)上運(yùn)轉(zhuǎn)起來的。因?yàn)樗麄儍H僅關(guān)注與數(shù)據(jù)處理的邏輯，因此可以極大的提高效率。而且應(yīng)該盡量不要限制編程語言，畢竟不同的公司甚至同一公司的不同部門使用的語言可能是千差萬別的。支持多語言無疑可以搶占更多的用戶。
2. Scale-out的設(shè)計(jì)：分布式系統(tǒng)天生就是scale-out的。
3. 無數(shù)據(jù)丟失：系統(tǒng)需要保證無數(shù)據(jù)丟失，這也是系統(tǒng)高可用性的保證。系統(tǒng)為了無數(shù)據(jù)丟失，需要在數(shù)據(jù)處理失敗的時候選擇另外的執(zhí)行路徑進(jìn)行replay（系統(tǒng)不是簡單的重新提交運(yùn)算，而是重新執(zhí)行調(diào)度，否則按照來源的call stack有可能使得系統(tǒng)永遠(yuǎn)都在相同的地方出同樣的錯誤）。
4. 容錯透明：用戶不會也不需要關(guān)心容錯。系統(tǒng)會自動處理容錯，調(diào)度并且管理資源，而這些行為對于運(yùn)行于其上的應(yīng)用來說都是透明的。
5. 數(shù)據(jù)持久化：為了保證高可用性和無數(shù)據(jù)丟失，數(shù)據(jù)持久化是無法躲避的問題。的確，數(shù)據(jù)持久化可能在低延時的系統(tǒng)中比較影響性能，但是這無法避免。當(dāng)然了，如果考慮到出錯情況比較少，在出錯的時候我們能夠忍受數(shù)據(jù)可以從頭replay，那么中間的運(yùn)算可以不進(jìn)行持久化。注意，這只有在持久化的成本要比計(jì)算的replay高的情況下有效。一般來說，計(jì)算的結(jié)果需要replica，當(dāng)然了，可以使用將數(shù)據(jù)replica到其他的節(jié)點(diǎn)的內(nèi)存中去（這又會占用集群的網(wǎng)絡(luò)帶寬）。
6. 超時設(shè)置：超時之所以在在這里被提出來，因?yàn)槌瑫r時間的大小設(shè)置需要重視，如果太短可以會誤殺正常運(yùn)行的計(jì)算，如果太長則不能快速的檢測錯誤。還有就是對于錯誤的快速發(fā)現(xiàn)可以這類系統(tǒng)的一個設(shè)計(jì)要點(diǎn)，畢竟，超時了才發(fā)現(xiàn)錯誤很多時候在時效性上是不可接受的。

原語設(shè)計(jì)

Hadoop定義了Map和Reduce，使得應(yīng)用者只需要實(shí)現(xiàn)MR就可以實(shí)現(xiàn)數(shù)據(jù)處理。而流式系統(tǒng)的特點(diǎn)，允許它們可以進(jìn)行更加具體一些的原語設(shè)計(jì)。流式的數(shù)據(jù)的特點(diǎn)就是數(shù)據(jù)時源源不斷進(jìn)入系統(tǒng)的，而這些數(shù)據(jù)的處理一般都需要幾個階段。拿普通的日志處理來說，我們可能僅僅關(guān)注Error的日志，那么系統(tǒng)的第一個計(jì)算邏輯就是進(jìn)行filer。接下來可能需要對這個日志進(jìn)行分段，分段后可能交給不同的規(guī)則處理器進(jìn)行處理。因此，數(shù)據(jù)處理一般是分階段的，可以說是一個有向無環(huán)圖，或者說是一個拓?fù)洹?shí)際上，Spark抽象出的運(yùn)算邏輯就是由RDD（Resilient Distributed Dataset）構(gòu)成DAG（Directed Acyclic Graph），而Storm則有Spout和Blot構(gòu)成Topology（拓?fù)洌?/p>

Spark的設(shè)計(jì)

Spark Streaming是將流式計(jì)算分解成一系列短小的批處理作業(yè)。這里的批處理引擎是Spark，也就是把Spark Streaming的輸入數(shù)據(jù)按照batch size（如1秒）分成一段一段的數(shù)據(jù)，每一段數(shù)據(jù)都轉(zhuǎn)換成Spark中的RDD，然后將Spark Streaming中對DStream的Transformation操作變?yōu)獒槍park中對RDD的Transformation操作，將RDD經(jīng)過操作變成中間結(jié)果保存在內(nèi)存中。整個流式計(jì)算根據(jù)業(yè)務(wù)的需求可以對中間的結(jié)果進(jìn)行疊加，或者存儲到外部設(shè)備。下圖顯示了Spark Streaming的整個流程。

WordCount的例子：

這個例子使用Scala寫的，一個簡單優(yōu)雅的函數(shù)式編程語言，同時也是基于JVM的后Java類語言。

Storm的設(shè)計(jì)

Storm將計(jì)算邏輯成為Topology，其中Spout是Topology的數(shù)據(jù)源，這個數(shù)據(jù)源可能是文件系統(tǒng)的某個日志，也可能是MessageQueue的某個消息隊(duì)列，也有可能是數(shù)據(jù)庫的某個表等等；Bolt負(fù)責(zé)數(shù)據(jù)的護(hù)理。Bolt有可能由另外兩個Bolt的join而來。

而Storm最核心的抽象Streaming就是連接Spout，Bolt以及Bolt與Bolt之間的數(shù)據(jù)流。而數(shù)據(jù)流的組成單位就是Tuple（元組），這個Tuple可能由多個Fields構(gòu)成，每個Field的含義都在Bolt的定義的時候制定。也就是說，對于一個Bolt來說，Tuple的格式是定義好的。

原語設(shè)計(jì)的要點(diǎn)

流式系統(tǒng)的原語設(shè)計(jì)，要關(guān)注一下幾點(diǎn)：

1. 如何定義計(jì)算拓?fù)洌阂奖闼惴ㄩ_發(fā)者開發(fā)算法與策略。最好的實(shí)現(xiàn)是定義一個算法與框架的交互方式，定義好算法的輸入結(jié)構(gòu)和算法的輸出結(jié)構(gòu)。然后拓?fù)淠軌蚪M合不同的算法來為用戶提供一個統(tǒng)一的服務(wù)。計(jì)算平臺最大的意義在于算法開發(fā)者不需要了解程序的運(yùn)行，并發(fā)的處理，高可用性的實(shí)現(xiàn)，只需要提供算法與計(jì)算邏輯即可以快速可靠的處理海量的數(shù)據(jù)。
2. 拓?fù)涞募虞d與啟動：對于每個節(jié)點(diǎn)來說，啟動時需要加載拓?fù)洌?jié)點(diǎn)需要其他的信息，比如上游的數(shù)據(jù)來源與下游的數(shù)據(jù)輸出。當(dāng)然了下游的數(shù)據(jù)輸出的拓?fù)湫畔⒖梢源鎯Φ絋uple中，對于數(shù)據(jù)需要放到那里去拓?fù)浔旧硎菬o狀態(tài)的。這就取決于具體的設(shè)計(jì)了。
3. 拓?fù)涞脑诰€更新：對于每個算法邏輯來說，更新是不可避免的，如何在不停止服務(wù)的情況下進(jìn)行更新是必要的。由于實(shí)現(xiàn)了架構(gòu)與算法的剝離，因此算法可以以一個單獨(dú)的個體進(jìn)行更新。可以操作如下：Master將算法實(shí)體保存到一個Worker可見的地方，比如HDFS或者是NFS或者ZK，然后通過心跳發(fā)送命令到拓?fù)洌負(fù)鋾簳r停止處理數(shù)據(jù)而加載新的算法實(shí)體，加載之后重新開始處理數(shù)據(jù)。數(shù)據(jù)一般都會放到buffer中，這個buffer可能是一個queue。但是從外界看來，拓?fù)鋵?shí)際上是一直處于服務(wù)狀態(tài)的。
4. 數(shù)據(jù)如何流動：流式系統(tǒng)最重要的抽象就是Streaming了。那么Steaming如何流動？實(shí)際上涉及到消息的傳遞和分發(fā)，數(shù)據(jù)如何從一個節(jié)點(diǎn)傳遞到另外一個節(jié)點(diǎn)，這是拓?fù)涠x的，具體實(shí)現(xiàn)可以參照第三小節(jié)。
5. 計(jì)算的終點(diǎn)及結(jié)果處理：流式計(jì)算的特點(diǎn)就是計(jì)算一直在進(jìn)行，流是源源不斷的流入到系統(tǒng)中的。但是對于每個數(shù)據(jù)單位來說它的處理結(jié)果是確定的，這個結(jié)果一般是需要返回調(diào)用者或者需要持久化的。比如處理一個時間段的交通違章，那么輸入的數(shù)據(jù)是一段時間的視頻監(jiān)控，輸出這是違章的信息，比如車牌，還有違章時刻的抓拍的圖片。這個數(shù)據(jù)要么返回調(diào)用者，由調(diào)用者負(fù)責(zé)數(shù)據(jù)的處理，包括持久化等。或者是拓?fù)渥詈蟮墓?jié)點(diǎn)將這些信息進(jìn)行持久化。系統(tǒng)需要對這些常見的case進(jìn)行指導(dǎo)性的說明，需要在Programmer Guide的sample中給出使用例子。

消息傳遞和分發(fā)

對于實(shí)現(xiàn)的邏輯來說，它們都是有向無環(huán)圖的一個節(jié)點(diǎn)，那么如何設(shè)計(jì)它們之間的消息傳遞呢？或者說數(shù)據(jù)如何流動的？因?yàn)閷τ诜植际较到y(tǒng)來說，我們不能假定整個運(yùn)算都是在同一個節(jié)點(diǎn)上（事實(shí)上，對于閉源軟件來說，這是可以的，比如就是滿足一個特定運(yùn)算下的計(jì)算，計(jì)算平臺也不需要做的那么通用，那么對于一個運(yùn)算邏輯讓他在一個節(jié)點(diǎn)完成也是可以了，畢竟節(jié)省了調(diào)度和網(wǎng)絡(luò)傳輸?shù)拈_銷）。或者說，對于一個通用的計(jì)算平臺來說，我們不能假定任何事情。

消息傳遞和分發(fā)是取決于系統(tǒng)的具體實(shí)現(xiàn)的。通過對比Storm和Spark，你就明白我為什么這么說了。

Spark的消息傳遞

對于Spark來說，數(shù)據(jù)流是在通過將用戶定義的一系列的RDD轉(zhuǎn)化成DAG圖，然后DAG Scheduler把這個DAG轉(zhuǎn)化成一個TaskSet，而這個TaskSet就可以向集群申請計(jì)算資源，集群把這個TaskSet部署到Worker中去運(yùn)算了。當(dāng)然了，對于開發(fā)者來說，他的任務(wù)是定義一些RDD，在RDD上做相應(yīng)的轉(zhuǎn)化動作，最后系統(tǒng)會將這一系列的RDD投放到Spark的集群中去運(yùn)行。

Storm的消息傳遞      

對于Storm來說，他的消息分發(fā)機(jī)制是在定義Topology的時候就顯式定義好的。也就是說，應(yīng)用程序的開發(fā)者需要清楚的定義各個Bolts之間的關(guān)系，下游的Bolt是以什么樣的方式獲取上游的Bolt發(fā)出的Tuple。Storm有六種消息分發(fā)模式：

1. Shuffle Grouping: 隨機(jī)分組，Storm會盡量把數(shù)據(jù)平均分發(fā)到下游Bolt中。
2. Fields Grouping：按字段分組， 比如按userid來分組， 具有同樣userid的tuple會被分到相同的Bolt。這個對于類似于WordCount這種應(yīng)用非常有幫助。
3. All Grouping： 廣播， 對于每一個Tuple， 所有的Bolts都會收到。這種分發(fā)模式要慎用，會造成資源的極大浪費(fèi)。
4. Global Grouping: 全局分組， 這個Tuple被分配到storm中的一個bolt的其中一個task。這個對于實(shí)現(xiàn)事務(wù)性的Topology非常有用。
5. Non Grouping: 不分組， 這個分組的意思是說stream不關(guān)心到底誰會收到它的tuple。目前這種分組和Shuffle grouping是一樣的效果， 有一點(diǎn)不同的是storm會把這個bolt放到這個bolt的訂閱者同一個線程里面去執(zhí)行。
6. Direct Grouping: 直接分組,  這是一種比較特別的分組方法，用這種分組意味著消息的發(fā)送者指定由消息接收者的哪個task處理這個消息。

消息傳遞要點(diǎn)

消息隊(duì)列現(xiàn)在是模塊之間通信的非常通用的解決方案了。消息隊(duì)列使得進(jìn)程間的通信可以跨越物理機(jī)，這對于分布式系統(tǒng)尤為重要，畢竟我們不能假定進(jìn)程究竟是部署在同一臺物理機(jī)上還是部署到不同的物理機(jī)上。RabbitMQ是應(yīng)用比較廣泛的MQ，關(guān)于RabbitMQ可以看我的一個專欄：RabbitMQ

提到MQ，不得不提的是ZeroMQ。ZeroMQ封裝了Socket，引用官方的說法： “ZMQ (以下 ZeroMQ 簡稱 ZMQ)是一個簡單好用的傳輸層，像框架一樣的一個 socket library，他使得 Socket 編程更加簡單、簡潔和性能更高。是一個消息處理隊(duì)列庫，可在多個線程、內(nèi)核和主機(jī)盒之間彈性伸縮。ZMQ 的明確目標(biāo)是“成為標(biāo)準(zhǔn)網(wǎng)絡(luò)協(xié)議棧的一部分，之后進(jìn)入 Linux 內(nèi)核”。現(xiàn)在還未看到它們的成功。但是，它無疑是極具前景的、并且是人們更加需要的“傳統(tǒng)”BSD 套接字之上的一層封裝。ZMQ 讓編寫高性能網(wǎng)絡(luò)應(yīng)用程序極為簡單和有趣。”

因此, ZeroMQ不是傳統(tǒng)意義上的MQ。它比較適用于節(jié)點(diǎn)之間和節(jié)點(diǎn)與Master之間的通信。Storm在0.8之前的Worker之間的通信就是通過ZeroMQ。但是為什么0.9就是用Netty替代了ZeroMQ呢？說替代不大合適，只是0.9的默認(rèn)的Worker之間的通信是使用了Netty，ZeroMQ還是支持的。Storm官方認(rèn)為ZeroMQ有以下缺點(diǎn)：

1. 不容易部署，尤其是在云環(huán)境下：以為ZMQ是以Ｃ寫的，因此它還是緊依賴于操作系統(tǒng)環(huán)境的。
2. 無法限制其內(nèi)存。通過JVM可以很容易的限制java所占用的內(nèi)存。但是ZMQ對于Storm來說是個黑盒似得存在。
3. Storm無法從ZMQ獲取信息。比如Storm無法知道當(dāng)前buffer中有多少數(shù)據(jù)為發(fā)送。

當(dāng)然了還有所謂的性能問題，具體可以訪問Netty作者的blog。結(jié)論就是Netty的性能比ZMQ（在默認(rèn)配置下）好兩倍。不知道所謂的ZMQ的默認(rèn)配置是什么。反正我對這個結(jié)果挺驚訝。當(dāng)然了，Netty使用Java實(shí)現(xiàn)的確方便了在Worker之間的通信加上授權(quán)和認(rèn)證機(jī)制。這個使用ZMQ的確是不太好做。

高可用性

HA是分布式系統(tǒng)的必要屬性。如果沒有HA，其實(shí)系統(tǒng)是不可用的。那么如果實(shí)現(xiàn)HA？對于Storm來說，它認(rèn)為Master節(jié)點(diǎn)Nimbus是無狀態(tài)的，無狀態(tài)意味著可以快速恢復(fù)，因此Nimbus并沒有實(shí)現(xiàn)HA（不知道以后的Nimbus是否會實(shí)現(xiàn)HA，實(shí)際上使用ZooKeeper實(shí)現(xiàn)節(jié)點(diǎn)的HA是開源領(lǐng)域的通用做法）。為什么說Nimbus是無狀態(tài)的呢？因?yàn)榧核械脑獢?shù)據(jù)都保存到了ZooKeeper（ZK）中。Nimbus定時從ZK獲取信息，并且通過向ZK寫信息來控制Worker。Worker也是通過從ZK中獲取信息，通過這種方式，Worker執(zhí)行從Nimbus傳遞過來的命令。

Storm的這種使用ZK的方式還是很值得借鑒的。

Spark是如何實(shí)現(xiàn)HA的？我的另外一篇文章分析過Spark的Master是怎么實(shí)現(xiàn)HA的：Spark技術(shù)內(nèi)幕：Master基于ZooKeeper的High Availability（HA）源碼實(shí)現(xiàn) 。

也是通過ZK的leader 選舉實(shí)現(xiàn)的。Spark使用了百行代碼的級別實(shí)現(xiàn)了Master的HA，由此可見ZK的功力。

除了這些Master的HA，還有每個Worker的HA。或者說Worker的HA說法不太準(zhǔn)確，因此對于集群里的工作節(jié)點(diǎn)來說，它可以非常容易失敗的。這里的HA可以說是如何讓W(xué)orker失敗后快速重啟，重新提供服務(wù)。實(shí)現(xiàn)方式也可以由很多種。一個簡單的方法就是使用一個容器（Container）啟動Worker并且監(jiān)控Worker的狀態(tài)，如果Worker異常退出，那么就重新啟動它。這個方法很簡單也很有效。

如果是節(jié)點(diǎn)宕機(jī)呢？上述方法肯定是不能用的。這種情況下Master會檢測到Worker的心跳超時，那么就會從資源池中把這個節(jié)點(diǎn)刪除。回到正題，宕機(jī)后的節(jié)點(diǎn)重啟涉及到了運(yùn)維方面的知識。對于一個集群來說，硬件宕機(jī)這種情況應(yīng)該需要統(tǒng)一的管理，也就是集群也可以由一個Master，維持每個節(jié)點(diǎn)的心跳來確定硬件的狀態(tài)。如果節(jié)點(diǎn)宕機(jī)，那么集群首先是重啟它。如果啟動失敗可能會通過電話或者短信或者郵件通知運(yùn)維人員。因此運(yùn)維人員為了保證集群的高可用性付出了很多的努力，尤其是大型互聯(lián)網(wǎng)公司的運(yùn)維人員，非常值得點(diǎn)贊。當(dāng)然了這個已經(jīng)不是Storm或者Spark所能涵蓋的了。

存儲模型與數(shù)據(jù)不丟失

其實(shí)，數(shù)據(jù)不丟失有時候和處理速度是矛盾的。為了數(shù)據(jù)不丟失就要進(jìn)行數(shù)據(jù)持久化，數(shù)據(jù)持久化意味著要寫硬盤，在固態(tài)硬盤還沒有成為標(biāo)配的今天，硬盤的IO速度永遠(yuǎn)是系統(tǒng)的痛點(diǎn)。當(dāng)然了可以在另外節(jié)點(diǎn)的內(nèi)存上進(jìn)行備份，但是這涉及到了集群的兩個稀缺資源：內(nèi)存和網(wǎng)絡(luò)。如果因?yàn)閭浞荻加昧舜罅康木W(wǎng)絡(luò)帶寬的話，那必將影響系統(tǒng)的性能，吞吐量。

當(dāng)然了，可以使用日志的方式。但是日志的話對于錯誤恢復(fù)的時間又是不太能接受的。流式計(jì)算系統(tǒng)的特點(diǎn)就是要快，如果錯誤恢復(fù)時間太長，那么可能不如直接replay來的快，而且系統(tǒng)設(shè)計(jì)還更為簡單。

其實(shí)如果不是為了追求100%的數(shù)據(jù)丟失，可以使用checkpoint的機(jī)制，允許一個時間窗口內(nèi)的數(shù)據(jù)丟失。

回到系統(tǒng)設(shè)計(jì)本身，實(shí)際上流式計(jì)算系統(tǒng)主要是為了離線和近線的機(jī)器學(xué)習(xí)和數(shù)據(jù)挖掘，因此肯定要保證數(shù)據(jù)的處理速度：至少系統(tǒng)可以處理一天的新增數(shù)據(jù)，否則數(shù)據(jù)堆積越來越大。因此即使有的數(shù)據(jù)處理丟失了數(shù)據(jù)，可以讓源頭重新發(fā)送數(shù)據(jù)。

還有另外一個話題，就是系統(tǒng)的元數(shù)據(jù)信心如何保存，因?yàn)橄到y(tǒng)的路由信息等需要是全局可見的，需要保存類似的這些數(shù)據(jù)以供集群查詢。當(dāng)然了Master節(jié)點(diǎn)保持了和所有節(jié)點(diǎn)的心跳，它完全可以保存這些數(shù)據(jù)，并且在心跳中可以返回這些數(shù)據(jù)。實(shí)際上HDFS的NameNode就是這么做的。HDFS的NN這種設(shè)計(jì)非常合理，為什么這么說？HDFS的元數(shù)據(jù)包含了非常多的數(shù)據(jù)：

1. 目錄文件樹結(jié)構(gòu)和文件與數(shù)據(jù)塊的對應(yīng)關(guān)系：會持久化到物理存儲中，文件名叫做fsimage。
2. DN與數(shù)據(jù)塊的對應(yīng)關(guān)系，即數(shù)據(jù)塊存儲在哪些DN中：在DN啟動時會上報(bào)到NN它所維護(hù)的數(shù)據(jù)塊。這個是動態(tài)建立的，不會持久化。因此，集群的啟動可能需要比較長的時間。

那么對于流式計(jì)算系統(tǒng)這種算得上輕量級的元數(shù)據(jù)來說，Master處理這些元數(shù)據(jù)實(shí)際上要簡單的多，當(dāng)然了，Master需要實(shí)現(xiàn)服務(wù)的HA和數(shù)據(jù)的HA。這些不是一個輕松的事情。實(shí)際上，可以采用ZooKeeper來保存系統(tǒng)的元數(shù)據(jù)。ZooKeeper使用一個目錄樹的結(jié)構(gòu)來保存集群的元數(shù)據(jù)。節(jié)點(diǎn)可以監(jiān)控感興趣的數(shù)據(jù)，如果數(shù)據(jù)有變化，那么節(jié)點(diǎn)會收到通知，然后就保證了系統(tǒng)級別的數(shù)據(jù)一致性。這點(diǎn)對于系統(tǒng)比較重要，因?yàn)楣?jié)點(diǎn)都是不穩(wěn)定的，因此系統(tǒng)的其他服務(wù)可能都會因?yàn)楣?jié)點(diǎn)失效而發(fā)生變化，這些都需要通知相關(guān)的節(jié)點(diǎn)更新器服務(wù)列表，保證了部分節(jié)點(diǎn)的失效并不會影響系統(tǒng)的整體的服務(wù)，從而也就實(shí)現(xiàn)了故障對于用戶的透明性。

如何與公司已有的生產(chǎn)環(huán)境進(jìn)行融合

包括Spark和Storm，在國內(nèi)著名的互聯(lián)網(wǎng)公司比如百度，淘寶和阿里巴巴都有應(yīng)用，但是它究竟貢獻(xiàn)了多少流量是不得而知的。我了解到的是實(shí)際上大部分的流量，尤其是核心流量還是走公司的老架構(gòu)的。著名的博主陳皓在微博上關(guān)于閉源軟件和開源軟件“特點(diǎn)”之爭算是引起了軒然大波，具體討論可以見知乎。之所以引用這個爭論也是為了切合本小節(jié)的主題：如何與公司已有的生產(chǎn)環(huán)境進(jìn)行融合。

雖然互聯(lián)網(wǎng)公司的產(chǎn)品迭代很快，但是公司的核心算法和架構(gòu)基本上改動不會那么多，因此公司不可能為了推動Storm和Spark這種開源產(chǎn)品而進(jìn)行大規(guī)模的重新開發(fā)。只有那么后起的項(xiàng)目，從零開始的項(xiàng)目，比如小規(guī)模的調(diào)研項(xiàng)目才可能用這些產(chǎn)品。當(dāng)然了開源產(chǎn)品首先是一個通用的平臺，但是通用有可能產(chǎn)生的代價就是不那么高效，對于某些特殊地方的不能根據(jù)特殊的應(yīng)用場景進(jìn)行優(yōu)化。如果對這個開源平臺進(jìn)行二次開發(fā)，使得性能方面滿足自己的需求，首先不管法務(wù)上的問題，對于自己私有版本和社區(qū)版本進(jìn)行merge也是個很大的challenge。就像現(xiàn)在很多公司對于Linux進(jìn)行了二次裁剪，開發(fā)自己需要的Linux一樣。都需要一些對于這些架構(gòu)非常熟悉，并且非常熟悉社區(qū)動態(tài)的人去做這些事情。而這些在互聯(lián)網(wǎng)公司，基本上是不可能的。因此大部分時候，都是自己做一個系統(tǒng)，去非常高效切合的去滿足自身的需求。

當(dāng)然了，開源社區(qū)的閃光點(diǎn)也會影響到閉源產(chǎn)品，閉源產(chǎn)品也會影響開源產(chǎn)品，這個相互影響是良性的，可以推動技術(shù)向前發(fā)展。

總結(jié)

Storm和Spark的設(shè)計(jì)，絕對不是一篇文章所能解決的。它里邊由非常多的哲學(xué)需要我們仔細(xì)去學(xué)習(xí)。它們可以說是我們進(jìn)行系統(tǒng)設(shè)計(jì)的良好的范例。本博客在接下來的半年會通過Spark的源碼來學(xué)習(xí)Spark的系統(tǒng)架構(gòu)。敬請期待！

原文鏈接： 從Storm和Spark Streaming學(xué)習(xí)流式實(shí)時分布式計(jì)算系統(tǒng)的設(shè)計(jì)要點(diǎn)（責(zé)編/魏偉）