前 言

近年來，隨著5G時代的到來以及物聯(lián)網(wǎng)和云計算的迅猛發(fā)展，人類社會逐漸步入了大數(shù)據(jù)時代。所謂大數(shù)據(jù)，是指所涉及的數(shù)據(jù)量規(guī)模巨大，無法通過人工在合理時間內(nèi)達(dá)到截取、管理、處理并整理成為人類所能解讀的信息。大數(shù)據(jù)在帶來發(fā)展機遇的同時，也帶來了新的挑戰(zhàn)，催生了新技術(shù)的發(fā)展和舊技術(shù)的革新。例如，不斷增長的數(shù)據(jù)規(guī)模和數(shù)據(jù)的動態(tài)快速產(chǎn)生要求必須采用分布式計算框架才能實現(xiàn)與之匹配的吞吐和實時性。

1.大數(shù)據(jù)計算基本概念

1.1 離線計算

大數(shù)據(jù)離線計算技術(shù)應(yīng)用于靜態(tài)數(shù)據(jù)的離線計算和處理，框架設(shè)計的初衷是為了解決大規(guī)模、非實時數(shù)據(jù)計算，更加關(guān)注整個計算框架的吞吐量。離線計算的數(shù)據(jù)量大且計算周期長，是在大量數(shù)據(jù)基礎(chǔ)上進(jìn)行復(fù)雜的批量運算。離線計算的數(shù)據(jù)是不再會發(fā)生變化，通常離線計算的任務(wù)都是定時的，使用場景一般式對時效性要求比較低的。

1.2 實時流式計算

實時流式計算，或者是實時計算，流式計算，在大數(shù)據(jù)領(lǐng)域都是差不多的概念。那么，到底什么是實時流式計算呢?谷歌大神Tyler Akidau在《the-world-beyond-batch-streaming-101》一文中提到過實時流式計算的三個特征：無限數(shù)據(jù)、無界數(shù)據(jù)處理、低延遲：

• 無限數(shù)據(jù)：指的是一種不斷增長的，基本上無限的數(shù)據(jù)集，這些通常被稱為“流數(shù)據(jù)”，而與之相對的是有限的數(shù)據(jù)集。
• 無界數(shù)據(jù)處理：是一種持續(xù)的數(shù)據(jù)處理模式，能夠通過處理引擎重復(fù)的去處理上面的無限數(shù)據(jù)，是能夠突破有限數(shù)據(jù)處理引擎的瓶頸。
• 低延遲：延遲是指數(shù)據(jù)從進(jìn)入系統(tǒng)到流出系統(tǒng)所用的時間，實時流式計算業(yè)務(wù)對延遲有較高要求，延遲越低，越能保證數(shù)據(jù)的實時性和有效性。

2.離線計算框架：大數(shù)據(jù)的主場

2.1 MapReduce計算框架

Hadoop是一個分布式系統(tǒng)架構(gòu)，由Apache基金會所開發(fā)，其核心主要包括兩個組件：HDFS和MapReduce，前者為海量存儲提供了存儲，而后者為海量的數(shù)據(jù)提供了計算。這里我們主要關(guān)注MapReduce。以下資料來源于Hadoop的官方說明文檔和論文。

MapReduce是一個使用簡易的軟件框架，基于它寫出來的應(yīng)用程序能夠運行在由上千個商用機器組成的大型集群上，并以一種可靠容錯的方式并行處理上T級別的數(shù)據(jù)集。將計算過程分為兩個階段，Map和Reduce，Map階段并行處理輸入的數(shù)據(jù)，Reduce階段對Map結(jié)果進(jìn)行匯總。

一個MapReduce作業(yè)通常會把輸入的數(shù)據(jù)集切分為若干獨立的數(shù)據(jù)塊，由Map任務(wù)以完全并行的方式處理它們。框架會對Map的輸出先進(jìn)行排序，然后把結(jié)果輸入給Reduce任務(wù)。通常作業(yè)的輸入和輸出都會被存儲在文件系統(tǒng)中。整個框架負(fù)責(zé)任務(wù)的調(diào)度和監(jiān)控，以及重新執(zhí)行已經(jīng)失敗的任務(wù)。

通常，MapReduce框架和分布式文件系統(tǒng)是運行在一組相同的節(jié)點上的，也就是說，計算節(jié)點和存儲節(jié)點通常在一起。這種配置允許框架在那些已經(jīng)存好數(shù)據(jù)的節(jié)點上高效地調(diào)度任務(wù)，這可以使整個集群的網(wǎng)絡(luò)帶寬被非常高效地利用。

MapReduce框架由一個單獨的master JobTracker 和每個集群節(jié)點一個slave TaskTracker共同組成。master負(fù)責(zé)調(diào)度構(gòu)成一個作業(yè)的所有任務(wù)，這些任務(wù)分布在不同的slave上，master監(jiān)控它們的執(zhí)行，重新執(zhí)行已經(jīng)失敗的任務(wù)。而slave僅負(fù)責(zé)執(zhí)行由master指派的任務(wù)。

應(yīng)用程序至少應(yīng)該指明輸入/輸出的路徑，并通過實現(xiàn)合適的接口或抽象類提供map和reduce函數(shù)。再加上其他作業(yè)的參數(shù)，就構(gòu)成了作業(yè)配置。然后，Hadoop的Job Client提交作業(yè)和配置信息給JobTracker，后者負(fù)責(zé)分發(fā)這些軟件和配置信息給slave、調(diào)度任務(wù)并監(jiān)控它們的執(zhí)行，同時提供狀態(tài)和診斷信息給Job Client。

MapReduce框架運轉(zhuǎn)在

應(yīng)用程序通常會通過提供map和reduce來實現(xiàn) Mapper和Reducer接口，它們組成作業(yè)的核心。map函數(shù)接受一個鍵值對，產(chǎn)生一組中間鍵值對。MapReduce框架會將map函數(shù)產(chǎn)生的中間鍵值對中鍵相同的值傳遞給一個reduce函數(shù)。reduce函數(shù)接受一個鍵，以及相關(guān)的一組值，將這組值進(jìn)行合并產(chǎn)生一組規(guī)模更小的值。如圖1所示，MapReduce的工作流程中，一切都是從最上方的user program開始的，user program鏈接了MapReduce庫，實現(xiàn)了最基本的Map函數(shù)和Reduce函數(shù)。圖中執(zhí)行的順序都用數(shù)字標(biāo)記了。

 

圖1 MapReduce的執(zhí)行流程

2.2 Spark計算框架

Spark基于MapReduce算法實現(xiàn)的離線計算，擁有Hadoop MapReduce所具有的優(yōu)點;但不同于MapReduce的是Job中間輸出結(jié)果可以保存在內(nèi)存中，從而不再需要讀寫HDFS，因此Spark能更好地適用于數(shù)據(jù)挖掘與機器學(xué)習(xí)等需要迭代的Map Reduce的算法。

Spark中一個主要的結(jié)構(gòu)是RDD(Resilient Distributed Datasets)，這是一種只讀的數(shù)據(jù)劃分，并且可以在丟失之后重建。它利用了Lineage的概念實現(xiàn)容錯，如果一個RDD丟失了，那么有足夠的信息支持RDD重建。RDD可以被認(rèn)為是提供了一種高度限制的共享內(nèi)存，但是這些限制可以使得自動容錯的開支變得很低。RDD使用Lineage的容錯機制，即每一個RDD都包含關(guān)于它是如何從其他RDD變換過來的以及如何重建某一塊數(shù)據(jù)的信息。RDD僅支持粗顆粒度變換，即僅記錄在單個塊上執(zhí)行的單個操作，然后創(chuàng)建某個RDD的變換序列存儲下來，當(dāng)數(shù)據(jù)丟失時，我們可以用變換序列來重新計算，恢復(fù)丟失的數(shù)據(jù)，以達(dá)到容錯的目的。

Spark中的應(yīng)用程序稱為驅(qū)動程序，這些驅(qū)動程序可實現(xiàn)在單一節(jié)點上執(zhí)行的操作或在一組節(jié)點上并行執(zhí)行的操作。驅(qū)動程序可以在數(shù)據(jù)集上執(zhí)行兩種類型的操作：動作和轉(zhuǎn)換。動作會在數(shù)據(jù)集上執(zhí)行一個計算，并向驅(qū)動程序返回一個值;而轉(zhuǎn)換會從現(xiàn)有數(shù)據(jù)集中創(chuàng)建一個新的數(shù)據(jù)集。動作的示例包括執(zhí)行一個Reduce操作以及在數(shù)據(jù)集上進(jìn)行迭代。轉(zhuǎn)換示例包括Map操作和Cache操作。

與Hadoop類似，Spark支持單節(jié)點集群或多節(jié)點集群。對于多節(jié)點操作，Spark依賴于Mesos集群管理器。Mesos為分布式應(yīng)用程序的資源共享和隔離提供了一個有效平臺，參考圖2。

 

圖2 Spark 依賴于Mesos集群管理器

2.3 Dryad計算框架

Dryad是構(gòu)建微軟云計算基礎(chǔ)設(shè)施的核心技術(shù)。編程模型相比MapReduce更具一般性——用有向無環(huán)圖(DAG)描述任務(wù)的執(zhí)行，其中用戶指定的程序是DAG圖的節(jié)點，數(shù)據(jù)傳輸?shù)耐ǖ朗沁叄赏ㄟ^文件、共享內(nèi)存或者傳輸控制協(xié)議(TCP)通道來傳遞數(shù)據(jù)，任務(wù)相當(dāng)于圖的生成器，可以合成任何圖，甚至在執(zhí)行的過程中這些圖也可以發(fā)生變化，以響應(yīng)計算過程中發(fā)生的事件。圖3給出了整個任務(wù)的處理流程。Dryad在容錯方面支持良好，底層的數(shù)據(jù)存儲支持?jǐn)?shù)據(jù)備份;在任務(wù)調(diào)度方面，Dryad的適用性更廣，不僅適用于云計算，在多核和多處理器以及異構(gòu)集群上同樣有良好的性能;在擴展性方面，可伸縮于各種規(guī)模的集群計算平臺，從單機多核計算機到由多臺計算機組成的集群，甚至擁有數(shù)千臺計算機的數(shù)據(jù)中心。Microsoft借助Dryad，在大數(shù)據(jù)處理方面也形成了完整的軟件棧，部署了分布式存系統(tǒng)Cosmos，提供DryadLINQ編程語言，使普通程序員可以輕易進(jìn)行大規(guī)模的分布式計算。

 

圖3 Dyrad計算框架的任務(wù)處理流程

3.實時流計算框架：大數(shù)據(jù)的未來

如果遇到時效性更敏感的業(yè)務(wù)需求，我們需要用到哪些實時計算引擎?目前有很多專業(yè)的實時流計算框架，較為知名的包括Apache Storm、Spark Streaming、LinkIn Samza、Apache Flink和Google MillWheel等，但是其中最主流的無疑是Storm、Spark Streaming、Flink和Samza。

3.1 Storm計算框架

Hadoop提供了Map和Reduce原語，使得對數(shù)據(jù)進(jìn)行批處理變得非常簡單和優(yōu)美。同樣，Storm也對數(shù)據(jù)的實時計算提供了簡單的Spout和Bolt原語。Storm集群表面上看和Hadoop集群非常像，但Hadoop上面運行的是MapReduce的Job，而Storm上面運行的是Topology，它們非常不一樣，比如一個MapReduce Job最終會結(jié)束，而一個Storm Topology永遠(yuǎn)運行。Storm的集群架構(gòu)如圖4所示。

 

圖4 Storm的集群架構(gòu)

在應(yīng)用Storm過程中會碰見Topology、Tuple、Spout、Bolt、流和流分組這些概念。其中Topology是一個實時應(yīng)用程序，Tuple是處理的基本消息單元，Spout是Topology的流的來源，是一個Topology中產(chǎn)生源數(shù)據(jù)流的組件，Topology中的所有處理邏輯都在Bolt中完成。一個流由無數(shù)個元組序列構(gòu)成，這些元組并行、分布式的被創(chuàng)建和執(zhí)行，流分組是用來定義一個Stream應(yīng)該如何分配數(shù)據(jù)給Bolts上的多個任務(wù)。

早期的Storm無法提供exactly once的語義支持，后期Storm引入了Trident高級原語，提供了exactly once的語義支持。然后提出了流計算中的反壓概念，指的是Storm中的一個拓?fù)涮幚頂?shù)據(jù)的速度小于數(shù)據(jù)流入的速度時的處理機制，通常來說，反壓出現(xiàn)的時候，數(shù)據(jù)會迅速累積，如果處理不當(dāng)，會導(dǎo)致資源耗盡甚至任務(wù)崩潰。這在流處理過程中非常常見，通常是由于源頭數(shù)據(jù)量突然急劇增加所導(dǎo)致的，比如電商的大促、節(jié)日活動等。新的Storm自動反壓機制通過監(jiān)控Bolt中的接收隊列的情況來實現(xiàn)，當(dāng)超過高水位值時，專門的線程會將反壓信息寫到ZooKeeper, ZooKeeper上的Watch會通知該拓?fù)涞乃蠾orker都進(jìn)入反壓狀態(tài)，最后Spout降低Tuple發(fā)送的速度。

3.2 Spark Streaming計算框架

Spark Streaming是Spark核心API的擴展，用于處理實時數(shù)據(jù)流。Spark Streaming處理的數(shù)據(jù)源可以是Kafka，F(xiàn)lume，Twitter，HDFS或者Kinesis，這些數(shù)據(jù)可以使用map，reduce，join，window方法進(jìn)行處轉(zhuǎn)換，還可以直接使用Spark內(nèi)置的機器學(xué)習(xí)算法，圖算法包來處理數(shù)據(jù)。最終處理后的數(shù)據(jù)可以存入HDFS，Database或者Dashboard中，數(shù)據(jù)庫。相比于Storm原生的實時處理框架，Spark Streaming是基于微批處理，微批處理是一種組織獨立數(shù)據(jù)操作的方法，術(shù)語中的微，更具體的說來，就是指在內(nèi)存中進(jìn)行處理。術(shù)語中的批處理指的是Spark Streaming中數(shù)據(jù)處理的單位是一批而不是一條，Spark會等采集的源頭數(shù)據(jù)累積到設(shè)置的間隔條件后，對數(shù)據(jù)進(jìn)行統(tǒng)一的批處理。這個間隔是Spark Streaming中的核心概念和關(guān)鍵參數(shù)，直接決定了Spark Streaming作業(yè)的數(shù)據(jù)處理延遲，當(dāng)然也決定著數(shù)據(jù)處理的吞吐量和性能。

Spark Streaming提供了一個叫做DStream的抽象概念，表示一段連續(xù)的數(shù)據(jù)流。在Spark Streaming內(nèi)部中，DStream實際上是由一系列連續(xù)的RDD組成的。每個RDD包含確定時間間隔內(nèi)的數(shù)據(jù)，這些離散的RDD連在一起，共同組成了對應(yīng)的DStream。Spark Streaming的架構(gòu)如下圖5所示。

 

圖5 Spark Streaming的架構(gòu)

3.3 Flink計算框架

Storm延遲低但是吞吐量小，Spark Streaming吞吐量大但是延遲高，那么是否有一種兼具低延遲和高吞吐量特點的流計算技術(shù)呢?答案是有的，就是Flink。實際上，F(xiàn)link于2008年作為柏林理工大學(xué)的一個研究性項目誕生，但是直到2015年以后才開始逐步得到認(rèn)可和接受，這和其自身的技術(shù)特點契合了大數(shù)據(jù)對低實時延遲、高吞吐、容錯、可靠性、靈活的窗口操作以及狀態(tài)管理等顯著特性分不開，當(dāng)然也和實時數(shù)據(jù)越來越得到重視分不開。阿里巴巴啟動了Blink項目，目標(biāo)是擴展、優(yōu)化、完善Flink，使其能夠應(yīng)用在阿里巴巴大規(guī)模實時計算場景。

Flink的整體結(jié)構(gòu)如下圖6所示。部署：Flink 支持本地運行(IDE 中直接運行程序)、能在獨立集群(Standalone模式)或者在被 YARN、Mesos、K8s 管理的集群上運行，也能部署在云上。內(nèi)核：Flink 的核心是分布式流式數(shù)據(jù)引擎，意味著數(shù)據(jù)以一次一個事件的形式被處理。API：包含了DataStream、DataSet、Table和SQL等API。庫：Flink還包括用于CEP(復(fù)雜事件處理)、機器學(xué)習(xí)、圖形處理等場景。

 

圖6 Flink的整體結(jié)構(gòu)

Flink的容錯機制核心是分布式數(shù)據(jù)流和狀態(tài)的快照，為了保證失敗時從錯誤中恢復(fù)，因此需要對數(shù)據(jù)對齊。Flink采用了單機性能十分優(yōu)異的RocksDB作為狀態(tài)的后端存儲，但單機是不可靠的，所以Flink還對將單機的狀態(tài)同步到HDFS上以保證狀態(tài)的可靠性。另外，對于從RocksDB到HDFS上checkpoint的同步，F(xiàn)link也支持增量的方式，能夠非常好地提高checkpoint的效率。Flink相比其他流計算技術(shù)的一個重要特性是支持基于Event Time的窗口操作。但是Event Time來自于源頭系統(tǒng)，網(wǎng)絡(luò)延遲、分布式處理以及源頭系統(tǒng)等各種原因?qū)е略搭^數(shù)據(jù)的事件時間可能是亂序的，即發(fā)生晚的事件反而比發(fā)生早的事件來得早，或者說某些事件會遲到。Flink參考Google的Cloud Dataflow，引入水印的概念來解決和衡量這種亂序的問題。并且在實時計算的某些場景，需要撤回之前的計算結(jié)果進(jìn)行，F(xiàn)link提供了撤回機制。

Storm是通過監(jiān)控process bolt中的接收隊列負(fù)載情況來處理反壓，如果超過高水位值，就將反壓信息寫到ZooKeeper，由ZooKeeper上的watch通知該拓?fù)涞乃衱orker都進(jìn)入反壓狀態(tài)，最后spout停止發(fā)送tuple來處理的。而Spark Streaming通過設(shè)置屬性“spark.streaming.backpressure.enabled”可以自動進(jìn)行反壓處理，它會動態(tài)控制數(shù)據(jù)接收速率來適配集群數(shù)據(jù)處理能力。對于Flink來說，不需要進(jìn)行任何的特殊設(shè)置，其本身的純數(shù)據(jù)流引擎可以非常優(yōu)雅地處理反壓問題。

3.4 Samza計算框架

Samza是Linkedin開源的分布式流處理框架，其架構(gòu)如圖8所示，由Kafka提供底層數(shù)據(jù)流，由YARN提供資源管理、任務(wù)分配等功能。圖7也給出了Samza的作業(yè)處理流程，即Samza客戶端負(fù)責(zé)將任務(wù)提交給YARN的資源管理器，后者分配相應(yīng)的資源完成任務(wù)的執(zhí)行。在每個容器中運行的流任務(wù)相對于Kafka是消息訂閱者，負(fù)責(zé)拉取消息并執(zhí)行相應(yīng)的邏輯。在可擴展性方 面，底層的Kafka通過Zookeeper實現(xiàn)了動態(tài)的集群水平擴展，可提供高吞吐、可水平擴展的消息隊列，YARN為Samza提供了分布式的環(huán)境和執(zhí)行容器，因此也很容易擴展;在容錯性方面，如果服務(wù)器出現(xiàn)故障，Samza和YARN將一起進(jìn)行任務(wù)的遷移、重啟和重新執(zhí)行，YARN還能提供任務(wù)調(diào)度、執(zhí)行狀態(tài)監(jiān)控等功能;在數(shù)據(jù)可靠性方面，Samza 按照Kafka中的消息分區(qū)進(jìn)行處理，分區(qū)內(nèi)保證消息有序，分區(qū)間并發(fā)執(zhí)行，Kafka將消息持久化到硬盤保證數(shù)據(jù)安全。另外，Samza還提供了對流數(shù)據(jù)狀態(tài)管理的支持。在需要記錄歷史數(shù)據(jù)的場景里，數(shù)據(jù)實時流動導(dǎo)致狀態(tài)管理難以實現(xiàn)，為此，Samza提供了一個內(nèi)建的鍵值數(shù)據(jù)庫用來存儲歷史數(shù)據(jù)。

 

圖7 Samza的整體架構(gòu)

4.總 結(jié)

大數(shù)據(jù)計算框架的應(yīng)用推進(jìn)了技術(shù)的發(fā)展和革新，目前業(yè)界在不斷提高大數(shù)據(jù)計算框架的吞吐量、實時性、可擴展性等特性以應(yīng)對日益增長的數(shù)據(jù)量和數(shù)據(jù)處理需求，大數(shù)據(jù)計算框架依然是現(xiàn)在以及未來一段時間內(nèi)的研究熱點。未來的發(fā)展趨勢是：隨著商業(yè)智能和計算廣告等領(lǐng)域的發(fā)展，更強調(diào)實時性的流計算框架將得到更加廣泛的關(guān)注。總之，應(yīng)用的推動和技術(shù)的進(jìn)步將會產(chǎn)生新的問題。作為大數(shù)據(jù)應(yīng)用的核心，對于挖掘數(shù)據(jù)價值起著重要作用的計算框架將會面臨更多的挑戰(zhàn)，亟待解決。本文參考了一些文獻(xiàn)和網(wǎng)絡(luò)資源，他們的觀點和技術(shù)對本文做出的貢獻(xiàn)表示感謝。

參考文獻(xiàn)

[1] 李川,鄂海紅,宋美娜.基于Storm的實時計算框架的研究與應(yīng)用[J].軟件,2014,35(10):16-20.

[2] https://izualzhy.cn/dataflow-reading

[3] https://juejin.im/post/5d49830cf265da03f3333b4c#heading-11

[4] Wenhong Tian, Yong Zhao, in Optimized Cloud Resource Management and Scheduling[M], 2015

[5] https://greeensy.github.io/2014/06/15/Batch-Computing/