一文讀懂大數據計算框架與平臺

　1.前言

　　計算機的基本工做就是處理數據，包括磁盤文件中的數據，經過網絡傳輸的數據流或數據包，數據庫中的結構化數據等。隨着互聯網、物聯網等技術獲得愈來愈普遍的應用，數據規模不斷增長，TB、PB量級成爲常態，對數據的處理已沒法由單臺計算機完成，而只能由多臺機器共同承擔計算任務。而在分佈式環境中進行大數據處理，除了與存儲系統打交道外，還涉及計算任務的分工，計算負荷的分配，計算機之間的數據遷移等工做，而且要考慮計算機或網絡發生故障時的數據安全，狀況要複雜得多。

　　舉一個簡單的例子，假設咱們要從銷售記錄中統計各類商品銷售額。在單機環境中，咱們只需把銷售記錄掃描一遍，對各商品的銷售額進行累加便可。若是銷售記錄存放在關係數據庫中，則更省事，執行一個SQL語句就能夠了。如今假定銷售記錄實在太多，須要設計出由多臺計算機來統計銷售額的方案。爲保證計算的正確、可靠、高效及方便，這個方案須要考慮下列問題：

　　如何爲每臺機器分配任務，是先按商品種類對銷售記錄分組，不一樣機器處理不一樣商品種類的銷售記錄，仍是隨機向各臺機器分發一部分銷售記錄進行統計，最後把各臺機器的統計結果按商品種類合併？

　　上述兩種方式都涉及數據的排序問題，應選擇哪一種排序算法？應該在哪臺機器上執行排序過程？

　　如何定義每臺機器處理的數據從哪裏來，處理結果到哪裏去？數據是主動發送，仍是接收方申請時才發送？若是是主動發送，接收方處理不過來怎麼辦？若是是申請時才發送，那發送方應該保存數據多久？

　　會不會任務分配不均，有的機器很快就處理完了，有的機器一直忙着？甚至，閒着的機器須要等忙着的機器處理完後才能開始執行？

　　若是增長一臺機器，它能不能減輕其餘機器的負荷，從而縮短任務執行時間？

　　若是一臺機器掛了，它沒有完成的任務該交給誰？會不會遺漏統計或重複統計？

　　統計過程當中，機器之間如何協調，是否須要專門的一臺機器指揮調度其餘機器？若是這臺機器掛了呢？

　　（可選）若是銷售記錄在源源不斷地增長，統計還沒執行完新記錄又來了，如何保證統計結果的準確性？能不能保證結果是實時更新的？再次統計時能不能避免大量重複計算？

　　（可選）能不能讓用戶執行一句SQL就能夠獲得結果？

　　上述問題中，除了第1個外，其他的都與具體任務無關，在其餘分佈式計算的場合也會遇到，並且解決起來都至關棘手。即便第1個問題中的分組、統計，在不少數據處理場合也會涉及，只是具體方式不一樣。若是能把這些問題的解決方案封裝到一個計算框架中，則可大大簡化這類應用程序的開發。

　　2004年先後，Google前後發表三篇論文分別介紹分佈式文件系統GFS、並行計算模型MapReduce、非關係數據存儲系統BigTable，第一次提出了針對大數據分佈式處理的可重用方案。在Google論文的啓發下，Yahoo的工程師DougCutting和MikeCafarella開發了Hadoop。在借鑑和改進Hadoop的基礎上，又前後誕生了數十種應用於分佈式環境的大數據計算框架。本文在參考業界慣例的基礎上，對這些框架按下列標準分類：

　　若是不涉及上面提出的第八、9兩個問題，則屬於批處理框架。批處理框架重點關心數據處理的吞吐量，又可分爲非迭代式和迭代式兩類，迭代式包括DAG（有向無環圖）、圖計算等模型。

　　若針對第8個問題提出來應對方案，則分兩種狀況：若是重點關心處理的實時性，則屬於流計算框架；若是側重於避免重複計算，則屬於增量計算框架。

　　若是重點關注的是第9個問題，則屬於交互式分析框架。

　　本文下面分別討論批處理、流計算、交互式分析三種類別的框架，而後簡要介紹大數據計算框架的一些發展趨勢。文章最後介紹這一領域的學習資料。

圖1.大數據計算框架全景圖

　　2.批處理框架

　　2.1.Hadoop

　　Hadoop最初主要包含分佈式文件系統HDFS和計算框架MapReduce兩部分，是從Nutch中獨立出來的項目。在2.0版本中，又把資源管理和任務調度功能從MapReduce中剝離造成YARN，使其餘框架也能夠像MapReduce那樣運行在Hadoop之上。與以前的分佈式計算框架相比，Hadoop隱藏了不少繁瑣的細節，如容錯、負載均衡等，更便於使用。

　　Hadoop也具備很強的橫向擴展能力，能夠很容易地把新計算機接入到集羣中參與計算。在開源社區的支持下，Hadoop不斷髮展完善，並集成了衆多優秀的產品如非關係數據庫HBase、數據倉庫Hive、數據處理工具Sqoop、機器學習算法庫Mahout、一致性服務軟件ZooKeeper、管理工具Ambari等，造成了相對完整的生態圈和分佈式計算事實上的標準。

圖2.Hadoop生態圈（刪減版）

　　MapReduce能夠理解爲把一堆雜亂無章的數據按照某種特徵歸併起來，而後處理並獲得最後的結果。基本處理步驟以下：

　　把輸入文件按照必定的標準分片，每一個分片對應一個map任務。通常狀況下，MapReduce和HDFS運行在同一組計算機上，也就是說，每臺計算機同時承擔存儲和計算任務，所以分片一般不涉及計算機之間的數據複製。

　　按照必定的規則把分片中的內容解析成鍵值對。一般選擇一種預約義的規則便可。

　　執行map任務，處理每一個鍵值對，輸出零個或多個鍵值對。

　　MapReduce獲取應用程序定義的分組方式，並按分組對map任務輸出的鍵值對排序。默認每一個鍵名一組。

　　待全部節點都執行完上述步驟後，MapReduce啓動Reduce任務。每一個分組對應一個Reduce任務。

　　執行reduce任務的進程經過網絡獲取指定組的全部鍵值對。

　　把鍵名相同的值合併爲列表。

　　執行reduce任務，處理每一個鍵對應的列表，輸出結果。

圖3.MapReduce處理過程

　　在上面的步驟中，應用程序主要負責設計map和reduce任務，其餘工做均由框架負責。在定義map任務輸出數據的方式時，鍵的選擇相當重要，除了影響結果的正確性外，也決定數據如何分組、排序、傳輸，以及執行reduce任務的計算機如何分工。前面提到的商品銷售統計的例子，可選擇商品種類爲鍵。MapReduce執行商品銷售統計的過程大體以下：

　　把銷售記錄分片，分配給多臺機器。

　　每條銷售記錄被解析成鍵值對，其中值爲銷售記錄的內容，鍵可忽略。

　　執行map任務，每條銷售記錄被轉換爲新的鍵值對，其中鍵爲商品種類，值爲該條記錄中商品的銷售額。

　　MapReduce把map任務生成的數據按商品種類排序。

　　待全部節點都完成排序後，MapReduce啓動reduce任務。每一個商品種類對應一個reduce任務。

　　執行reduce任務的進程經過網絡獲取指定商品種類的各次銷售額。

　　MapReduce把同一種商品下的各次銷售額合併到列表中。

　　執行reduce任務，累加各次銷售額，獲得該種商品的總銷售額。

　　上面的過程還有優化的空間。在傳輸各類商品每次的銷售額數據前，可先在map端對各類商品的銷售額進行小計，由此可大大減小網絡傳輸的負荷。MapReduce經過一個可選的combine任務支持該類型的優化。

　　2.2.DAG模型

　　如今假設咱們的目標更進一步，但願知道銷售得最好的前10種商品。咱們能夠分兩個環節來計算：

　　統計各類商品的銷售額。經過MapReduce實現，這在前面已經討論過。

　　對商品種類按銷售額排名。能夠經過一個排序過程完成。假定商品種類很是多，須要經過多臺計算機來加快計算速度的話，咱們能夠用另外一個MapReduce過程來實現，其基本思路是把map和reduce分別看成小組賽和決賽，先計算各分片的前10名，彙總後再計算總排行榜的前10名。

　　從上面的例子能夠看出，經過多個MapReduce的組合，能夠表達複雜的計算問題。不過，組合過程須要人工設計，比較麻煩。另外，每一個階段都須要全部的計算機同步，影響了執行效率。

　　爲克服上述問題，業界提出了DAG（有向無環圖）計算模型，其核心思想是把任務在內部分解爲若干存在前後順序的子任務，由此可更靈活地表達各類複雜的依賴關係。MicrosoftDryad、GoogleFlumeJava、ApacheTez是最先出現的DAG模型。Dryad定義了串接、全鏈接、融合等若干簡單的DAG模型，經過組合這些簡單結構來描述複雜的任務，FlumeJava、Tez則經過組合若干MapReduce造成DAG任務。

圖4.MapReduce（左）與Tez（右）

執行復雜任務時對比

　　MapReduce的另外一個不足之處是使用磁盤存儲中間結果，嚴重影響了系統的性能，這在機器學習等須要迭代計算的場合更爲明顯。加州大學伯克利分校AMP實驗室開發的Spark克服了上述問題。Spark對早期的DAG模型做了改進，提出了基於內存的分佈式存儲抽象模型RDD（ResilientDistributedDatasets，可恢復分佈式數據集），把中間數據有選擇地加載並駐留到內存中，減小磁盤IO開銷。與Hadoop相比，Spark基於內存的運算要快100倍以上，基於磁盤的運算也要快10倍以上。

圖5.MapReduce與Spark中間結果

保存方式對比

　　Spark爲RDD提供了豐富的操做方法，其中map、filter、flatMap、sample、groupByKey、reduceByKey、union、join、cogroup、mapValues、sort、partionBy用於執行數據轉換，生成新的RDD，而count、collect、reduce、lookup、save用於收集或輸出計算結果。如前面統計商品銷售額的例子，在Spark中只須要調用map和reduceByKey兩個轉換操做就能夠實現，整個程序包括加載銷售記錄和保存統計結果在內也只須要寥寥幾行代碼，而且支持Java、Scala、Python、R等多種開發語言，比MapReduce編程要方便得多。下圖說明reduceByKey的內部實現。

圖6.RDDreduceByKey內部實現

　　RDD因爲把數據存放在內存中而不是磁盤上，所以須要比Hadoop更多地考慮容錯問題。分佈式數據集的容錯有兩種方式：數據檢查點和記錄數據的更新。處理海量數據時，數據檢查點操做成本很高，所以Spark默認選擇記錄更新的方式。不過若是更新粒度太細太多，記錄更新成本也不低。所以，RDD只支持粗粒度轉換，即只記錄單個塊上執行的單個操做，而後將建立RDD的一系列變換序列記錄下來，相似於數據庫中的日誌。

　　當RDD的部分分區數據丟失時，Spark根據以前記錄的演變過程從新運算，恢復丟失的數據分區。Spark生態圈的另外一項目Alluxio（原名Tachyon）也採用相似的思路，使數據寫入速度比HDFS有數量級的提高。

　　下面總結Spark對MapReduce的改進：

　　MapReduce抽象層次低，須要手工編寫代碼完成；Spark基於RDD抽象，使數據處理邏輯的代碼很是簡短。

　　MapReduce只提供了map和reduce兩個操做，表達力欠缺；Spark提供了不少轉換和動做，不少關係數據庫中常見的操做如JOIN、GROUPBY已經在RDD中實現。

　　MapReduce中，只有map和reduce兩個階段，複雜的計算須要大量的組合，而且由開發者本身定義組合方式；Spark中，RDD能夠連續執行多個轉換操做，若是這些操做對應的RDD分區不變的話，還能夠放在同一個任務中執行。

　　MapReduce處理邏輯隱藏在代碼中，不直觀；Spark代碼不包含操做細節，邏輯更清晰。

　　MapReduce中間結果放在HDFS中；Spark中間結果放在內存中，內存放不下時才寫入本地磁盤而不是HDFS，這顯著提升了性能，特別是在迭代式數據處理的場合。

　　MapReduce中，reduce任務須要等待全部map任務完成後才能夠開始；在Spark中，分區相同的轉換構成流水線放到同一個任務中運行。

　　3.流計算框架

　　3.1.流計算概述

　　在大數據時代，數據一般都是持續不斷動態產生的。在不少場合，數據須要在很是短的時間內獲得處理，而且還要考慮容錯、擁塞控制等問題，避免數據遺漏或重複計算。流計算框架則是針對這一類問題的解決方案。流計算框架通常採用DAG（有向無環圖）模型。圖中的節點分爲兩類：一類是數據的輸入節點，負責與外界交互而向系統提供數據；另外一類是數據的計算節點，負責完成某種處理功能如過濾、累加、合併等。從外部系統不斷傳入的實時數據則流經這些節點，把它們串接起來。若是把數據流比做水的話，輸入節點比如是噴頭，源源不斷地出水，計算節點則至關於水管的轉接口。以下圖所示。

圖7.流計算DAG模型示意圖

　　爲提升併發性，每個計算節點對應的數據處理功能被分配到多個任務（相同或不一樣計算機上的線程）。在設計DAG時，須要考慮如何把待處理的數據分發到下游計算節點對應的各個任務，這在實時計算中稱爲分組（Grouping）。最簡單的方案是爲每一個任務複製一份，不過這樣效率很低，更好的方式是每一個任務處理數據的不一樣部分。隨機分組能達到負載均衡的效果，應優先考慮。不過在執行累加、數據關聯等操做時，須要保證同一屬性的數據被固定分發到對應的任務，這時應採用定向分組。在某些狀況下，還須要自定義分組方案。

圖8.流計算分組

　　因爲應用場合的普遍性，目前市面上已經有很多流計算平臺，包括GoogleMillWheel、TwitterHeron和Apache項目Storm、Samza、S四、Flink、Apex、Gearpump。

　　3.2.Storm及Trident

　　在流計算框架中，目前人氣最高，應用最普遍的要數Storm。這是因爲Storm具備簡單的編程模型，且支持Java、Ruby、Python等多種開發語言。Storm也具備良好的性能，在多節點集羣上每秒能夠處理上百萬條消息。Storm在容錯方面也設計得很優雅。下面介紹Storm確保消息可靠性的思路。

　　在DAG模型中，確保消息可靠的難點在於，原始數據被當前的計算節點成功處理後，還不能被丟棄，由於它生成的數據仍然可能在後續的計算節點上處理失敗，須要由該消息從新生成。而若是要對消息在各個計算節點的處理狀況都做跟蹤記錄的話，則會消耗大量資源。

　　Storm的解決思路，是爲每條消息分派一個ID做爲惟一性標識，並在消息中包含原始輸入消息的ID。同時用一個響應中心（Acker）維護每條原始輸入消息的狀態，狀態的初值爲該原始輸入消息的ID。每一個計算節點成功執行後，則把輸入和輸出消息的ID進行異或，再異或對應的原始輸入消息的狀態。因爲每條消息在生成和處理時分別被異或一次，則成功執行後全部消息均被異或兩次，對應的原始輸入消息的狀態爲0。所以當狀態爲0後可安全清除原始輸入消息的內容，而若是超過指定時間間隔後狀態仍不爲0，則認爲處理該消息的某個環節出了問題，須要從新執行。

圖9.Storm保證消息可靠性過程示意圖

　　Storm還實現了更高層次的抽象框架Trident。Trident以微批處理的方式處理數據流，好比每次處理100條記錄。Trident提供了過濾、分組、鏈接、窗口操做、聚合、狀態管理等操做，支持跨批次進行聚合處理，並對執行過程進行優化，包括多個操做的合併、數據傳輸前的本地聚合等。以微批處理方式處理數據流的框架還有SparkStreaming。

圖10.實時流處理與微批處理比較

　　下面是Storm、Trident與另外幾種流計算框架的對比：

　　4.交互式分析框架

　　4.1.概述

　　在解決了大數據的可靠存儲和高效計算後，如何爲數據分析人員提供便利日益受到關注，而最便利的分析方式莫過於交互式查詢。這幾年交互式分析技術發展迅速，目前這一領域知名的平臺有十餘個，包括Google開發的Dremel和PowerDrill，Facebook開發的Presto，Hadoop服務商Cloudera和HortonWorks分別開發的Impala和Stinger，以及Apache項目Hive、Drill、Tajo、Kylin、MRQL等。

　　一些批處理和流計算平臺如Spark和Flink也分別內置了交互式分析框架。因爲SQL已被業界普遍接受，目前的交互式分析框架都支持用相似SQL的語言進行查詢。早期的交互式分析平臺創建在Hadoop的基礎上，被稱做SQL-on-Hadoop。後來的分析平臺改用Spark、Storm等引擎，不過SQL-on-Hadoop的稱呼仍是沿用了下來。SQL-on-Hadoop也指爲分佈式數據存儲提供SQL查詢功能。

　　4.2.Hive

　　ApacheHive是最先出現的架構在Hadoop基礎之上的大規模數據倉庫，由Facebook設計並開源。Hive的基本思想是，經過定義模式信息，把HDFS中的文件組織成相似傳統數據庫的存儲系統。Hive保持着Hadoop所提供的可擴展性和靈活性。Hive支持熟悉的關係數據庫概念，好比表、列和分區，包含對非結構化數據必定程度的SQL支持。它支持全部主要的原語類型（如整數、浮點數、字符串）和複雜類型（如字典、列表、結構）。它還支持使用相似SQL的聲明性語言HiveQueryLanguage(HiveQL)表達的查詢，任何熟悉SQL的人都很容易理解它。HiveQL被編譯爲MapReduce過程執行。下圖說明如何經過MapReduce實現JOIN和GROUPBY。

(1)實現JOIN

(2)實現GROUPBY

圖11.部分HiveQL操做的實現方式

　　Hive與傳統關係數據庫對好比下：

　　Hive的主要弱點是因爲創建在MapReduce的基礎上，性能受到限制。不少交互式分析平臺基於對Hive的改進和擴展，包括Stinger、Presto、Kylin等。其中Kylin是中國團隊提交到Apache上的項目，其不同凡響的地方是提供多維分析（OLAP）能力。Kylin對多維分析可能用到的度量進行預計算，供查詢時直接訪問，由此提供快速查詢和高併發能力。Kylin在eBay、百度、京東、網易、美團均有應用。

　　4.3.SQL引擎Calcite

　　對於交互式分析，SQL查詢引擎的優劣對性能的影響舉足輕重。Spark開發了本身的查詢引擎Catalyst，而包括Hive、Drill、Kylin、Flink在內的不少交互式分析平臺及數據倉庫使用Calcite（原名optiq）做爲SQL引擎。Calcite是一個Apache孵化項目，其建立者JulianHyde曾是Oracle數據庫SQL引擎的主要開發者。Calcite具備下列幾個技術特色：

　　支持標準SQL語言。

　　支持OLAP。

　　支持對流數據的查詢。

　　獨立於編程語言和數據源，能夠支持不一樣的前端和後端。

　　支持關係代數、可定製的邏輯規劃規則和基於成本模型優化的查詢引擎。

　　支持物化視圖（materializedview）的管理。

　　因爲分佈式場景遠比傳統的數據存儲環境更復雜，Calcite和Catalyst都還處於向Oracle、MySQL等經典關係數據庫引擎學習的階段，在性能優化的道路上還有很長的路要走。

　　5.其餘類型的框架

　　除了上面介紹的幾種類型的框架外，還有一些目前還不太熱門但具備重要潛力的框架類型。圖計算是DAG以外的另外一種迭代式計算模型，它以圖論爲基礎對現實世界建模和計算，擅長表達數據之間的關聯性，適用於PageRank計算、社交網絡分析、推薦系統及機器學習。這一類框架有GooglePregel、ApacheGiraph、ApacheHama、PowerGraph、，其中PowerGraph是這一領域目前最傑出的表明。不少圖數據庫也內置圖計算框架。

　　另外一類是增量計算框架，探討如何只對部分新增數據進行計算來極大提高計算過程的效率，可應用到數據增量或週期性更新的場合。這一類框架包括GooglePercolator、MicrosoftKineograph、阿里Galaxy等。

　　另外還有像ApacheIgnite、ApacheGeode（GemFire的開源版本）這樣的高性能事務處理框架。

　　6.總結與展望

　　從Hadoop橫空出世到如今10餘年的時間中，大數據分佈式計算技術獲得了迅猛發展。不過因爲歷史尚短，這方面的技術遠未成熟。各類框架都還在不斷改進，並相互競爭。

　　性能優化毫無疑問是大數據計算框架改進的重點方向之一。而性能的提升很大程度上取決於內存的有效利用。這包括前面提到的內存計算，現已在各類類型的框架中普遍採用。內存資源的分配管理對性能也有重要影響，JVM垃圾回收在給開發人員帶來便利的同時，也制約了內存的有效利用。另外，Java的對象建立及序列化也比較浪費資源。在內存優化方面作足功夫的表明是Flink。出於性能方面的考慮，Flink不少組件自行管理內存，無需依賴JVM垃圾回收機制。Flink還用到開闢內存池、用二進制數據代替對象、量身定製序列化、定製緩存友好的算法等優化手段。Flink還在任務的執行方面進行優化，包括多階段並行執行和增量迭代。

　　擁抱機器學習和人工智能也是大數據計算的潮流之一。Spark和Flink分別推出機器學習庫SparkML和FlinkML。更多的平臺在第三方大數據計算框架上提供機器學習，如Mahout、Oryx及一干Apache孵化項目SystemML、HiveMall、PredictionIO、SAMOA、MADLib。這些機器學習平臺通常都同時支持多個計算框架，如Mahout同時以Spark、Flink、H2O爲引擎，SAMOA則使用S四、Storm、Samza。在深度學習掀起熱潮後，又有社區探索把深度學習框架與現有分佈式計算框架結合起來，這樣的項目有SparkNet、CaffeonSpark、TensorFrames等。

　　在同一平臺上支持多種框架也是發展趨勢之一，尤爲對於那些開發實力較爲雄厚的社區。Spark以批處理模型爲核心，實現了交互式分析框架SparkSQL、流計算框架SparkStreaming（及正在實現的StructuredStreaming）、圖計算框架GraphX、機器學習庫SparkML。而Flink在提供低延遲的流計算的同時，批處理、關係計算、圖計算、機器學習，一個也沒落下，目標直奔大數據通用計算平臺。Google的BEAM（意爲Batch+strEAM）則試圖把Spark、Flink、Apex這樣的計算框架歸入本身制定的標準之下，很有號令江湖之意。

圖12.BEAM的統一模型

　　7.學習資料

　　最後介紹一下大數據計算方面的學習資料。入門前的瞭解、知識面的拓展及知識的零散積累靠長期訪問相關的網站、論壇、微信訂閱號，問題解答則靠對搜索引擎的熟練駕馭。須要指出的是，網上的內容良萎不齊，不少資料是過期的，以訛傳訛也是常有的事，要注意鑑別。

　　論壇首推知乎、Quora、StackOverflow，運氣好的話開發者親自給你解答。其餘值得關注的網站或論壇包括煉數成金、人大經濟論壇、CSDN、博客園、雲棲社區、360大數據、推酷、伯樂在線、小象學院等。微信訂閱號中，InfoQ是最權威的，其餘還有THU數據派、大數據雜談、CSDN大數據、數據猿、Hadoop技術博文等，各人根據偏好取捨。

　　若要進行系統的學習，則首先應參考官方網站文檔。很多大數據平臺的官方文檔內容都比較詳實，賽過多數教材。另外，官方文檔與產品一般同步更新，這個優點是其餘資料沒法作到的。不過要說可讀性，書籍或視頻教程要強得多。視頻資料能夠從上文提到的部分網站論壇下載。

　　書籍方面，國外O\'Reilly、Manning兩家出版社在大數據領域出版了很多優秀書籍，特別是Manning的InAction系列和O\'Reilly的DefinitiveGuide系列。前者側重提升動手能力，後者則知識比較全面。InAction和DefinitiveGuide系列的書籍不少已翻譯爲中文，通常分別譯爲xxx實戰、xxx權威指南。另一家出版社Packt也值得關注。Packt的書比較薄，適合入門。至於中文原創書籍，推薦張俊林的《大數據日知錄》，該書是對大數據存儲和處理技術的全面梳理，系統性強。其餘書籍不逐一點評，若想購買或閱讀可參考豆瓣對該書的評分。