1. 前言
計算機的基本工作就是處理資料，包括磁盤檔案中的資料，通過網絡傳輸的資料流或資料包，資料庫中的結構化資料等。随着網際網路、物聯網等技術得到越來越廣泛的應用，資料規模不斷增加，tb、pb量級成為常态，對資料的處理已無法由單台計算機完成，而隻能由多台機器共同承擔計算任務。而在分布式環境中進行大資料處理，除了與存儲系統打交道外，還涉及計算任務的分工，計算負荷的配置設定，計算機之間的資料遷移等工作，并且要考慮計算機或網絡發生故障時的資料安全，情況要複雜得多。
舉一個簡單的例子，假設我們要從銷售記錄中統計各種商品銷售額。在單機環境中，我們隻需把銷售記錄掃描一遍，對各商品的銷售額進行累加即可。如果銷售記錄存放在關系資料庫中，則更省事，執行一個sql語句就可以了。現在假定銷售記錄實在太多，需要設計出由多台計算機來統計銷售額的方案。為保證計算的正确、可靠、高效及友善，這個方案需要考慮下列問題：
如何為每台機器配置設定任務，是先按商品種類對銷售記錄分組，不同機器處理不同商品種類的銷售記錄，還是随機向各台機器分發一部分銷售記錄進行統計，最後把各台機器的統計結果按商品種類合并？
上述兩種方式都涉及資料的排序問題，應選擇哪種排序算法？應該在哪台機器上執行排序過程？
如何定義每台機器處理的資料從哪裡來，處理結果到哪裡去？資料是主動發送，還是接收方申請時才發送？如果是主動發送，接收方處理不過來怎麼辦？如果是申請時才發送，那發送方應該儲存資料多久？
會不會任務配置設定不均，有的機器很快就處理完了，有的機器一直忙着？甚至，閑着的機器需要等忙着的機器處理完後才能開始執行？
如果增加一台機器，它能不能減輕其他機器的負荷，進而縮短任務執行時間？
如果一台機器挂了，它沒有完成的任務該交給誰？會不會遺漏統計或重複統計？
統計過程中，機器之間如何協調，是否需要專門的一台機器指揮排程其他機器？如果這台機器挂了呢？
（可選）如果銷售記錄在源源不斷地增加，統計還沒執行完新記錄又來了，如何保證統計結果的準确性？能不能保證結果是實時更新的？再次統計時能不能避免大量重複計算？
（可選）能不能讓使用者執行一句sql就可以得到結果？
上述問題中，除了第1個外，其餘的都與具體任務無關，在其他分布式計算的場合也會遇到，而且解決起來都相當棘手。即使第1個問題中的分組、統計，在很多資料處理場合也會涉及，隻是具體方式不同。如果能把這些問題的解決方案封裝到一個計算架構中，則可大大簡化這類應用程式的開發。
2004年前後，google先後發表三篇論文分别介紹分布式檔案系統gfs、并行計算模型mapreduce、非關系資料存儲系統bigtable，第一次提出了針對大資料分布式處理的可重用方案。在google論文的啟發下，yahoo的工程師doug cutting和mike cafarella開發了hadoop。在借鑒和改進hadoop的基礎上，又先後誕生了數十種應用于分布式環境的大資料計算架構。本文在參考業界慣例的基礎上，對這些架構按下列标準分類：
如果不涉及上面提出的第8、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（有向無環圖）計算模型，其核心思想是把任務在内部分解為若幹存在先後順序的子任務，由此可更靈活地表達各種複雜的依賴關系。microsoft dryad、google flumejava、apache tez是最早出現的dag模型。dryad定義了串接、全連接配接、融合等若幹簡單的dag模型，通過組合這些簡單結構來描述複雜的任務，flumejava、tez則通過組合若幹mapreduce形成dag任務。
圖4. mapreduce（左）與tez（右）
執行複雜任務時對比
mapreduce的另一個不足之處是使用磁盤存儲中間結果，嚴重影響了系統的性能，這在機器學習等需要疊代計算的場合更為明顯。加州大學伯克利分校amp實驗室開發的spark克服了上述問題。spark對早期的dag模型作了改進，提出了基于記憶體的分布式存儲抽象模型rdd（resilient distributed datasets，可恢複分布式資料集），把中間資料有選擇地加載并駐留到記憶體中，減少磁盤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. rdd reducebykey内部實作
rdd由于把資料存放在記憶體中而不是磁盤上，是以需要比hadoop更多地考慮容錯問題。分布式資料集的容錯有兩種方式：資料檢查點和記錄資料的更新。處理海量資料時，資料檢查點操作成本很高， 是以spark預設選擇記錄更新的方式。不過如果更新粒度太細太多，記錄更新成本也不低。是以，rdd隻支援粗粒度轉換，即隻記錄單個塊上執行的單個操作，然後将建立rdd的一系列變換序列記錄下來，類似于資料庫中的日志。
當rdd的部分分區資料丢失時，spark根據之前記錄的演變過程重新運算，恢複丢失的資料分區。spark生态圈的另一項目alluxio（原名tachyon）也采用類似的思路，使資料寫入速度比hdfs有數量級的提升。
下面總結spark對mapreduce的改進：
mapreduce抽象層次低，需要手工編寫代碼完成；spark基于rdd抽象，使資料處理邏輯的代碼非常簡短。
mapreduce隻提供了map和reduce兩個操作，表達力欠缺；spark提供了很多轉換和動作，很多關系資料庫中常見的操作如join、group by已經在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. 流計算分組
由于應用場合的廣泛性，目前市面上已經有不少流計算平台，包括google millwheel、twitter heron和apache項目storm、samza、s4、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提供了過濾、分組、連接配接、視窗操作、聚合、狀态管理等操作，支援跨批次進行聚合處理，并對執行過程進行優化，包括多個操作的合并、資料傳輸前的本地聚合等。以微批處理方式處理資料流的架構還有spark streaming。
(1) 實時流處理
(2) 微批處理
圖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
apache hive是最早出現的架構在hadoop基礎之上的大規模資料倉庫，由facebook設計并開源。hive的基本思想是，通過定義模式資訊，把hdfs中的檔案組織成類似傳統資料庫的存儲系統。hive 保持着 hadoop 所提供的可擴充性和靈活性。hive支援熟悉的關系資料庫概念，比如表、列和分區，包含對非結構化資料一定程度的 sql 支援。它支援所有主要的原語類型（如整數、浮點數、字元串）和複雜類型（如字典、清單、結構）。它還支援使用類似 sql 的聲明性語言 hive query language (hiveql) 表達的查詢，任何熟悉 sql 的人都很容易了解它。hiveql被編譯為mapreduce過程執行。下圖說明如何通過mapreduce實作join和group by。
(1) 實作join
(2) 實作group by
圖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孵化項目，其建立者julian hyde曾是oracle資料庫sql引擎的主要開發者。calcite具有下列幾個技術特點：
支援标準sql語言。
支援olap。
支援對流資料的查詢。
獨立于程式設計語言和資料源，可以支援不同的前端和後端。
支援關系代數、可定制的邏輯規劃規則和基于成本模型優化的查詢引擎。
支援物化視圖（materialized view）的管理。
由于分布式場景遠比傳統的資料存儲環境更複雜，calcite和catalyst都還處于向oracle、mysql等經典關系資料庫引擎學習的階段，在性能優化的道路上還有很長的路要走。
5. 其他類型的架構
除了上面介紹的幾種類型的架構外，還有一些目前還不太熱門但具有重要潛力的架構類型。圖計算是dag之外的另一種疊代式計算模型，它以圖論為基礎對現實世界模組化和計算，擅長表達資料之間的關聯性，适用于pagerank計算、社交網絡分析、推薦系統及機器學習。這一類架構有google pregel、apache giraph、apache hama、powergraph、，其中powergraph是這一領域目前最傑出的代表。很多圖資料庫也内置圖計算架構。
另一類是增量計算架構，探讨如何隻對部分新增資料進行計算來極大提升計算過程的效率，可應用到資料增量或周期性更新的場合。這一類架構包括google percolator、microsoft kineograph、阿裡galaxy等。
另外還有像apache ignite、apache geode（gemfire的開源版本）這樣的高性能事務處理架構。
6. 總結與展望
從hadoop橫空出世到現在10餘年的時間中，大資料分布式計算技術得到了迅猛發展。不過由于曆史尚短，這方面的技術遠未成熟。各種架構都還在不斷改進，并互相競争。
性能優化毫無疑問是大資料計算架構改進的重點方向之一。而性能的提高很大程度上取決于記憶體的有效利用。這包括前面提到的記憶體計算，現已在各種類型的架構中廣泛采用。記憶體資源的配置設定管理對性能也有重要影響，jvm垃圾回收在給開發人員帶來便利的同時，也制約了記憶體的有效利用。另外，java的對象建立及序列化也比較浪費資源。在記憶體優化方面做足功夫的代表是flink。出于性能方面的考慮，flink很多元件自行管理記憶體，無需依賴jvm垃圾回收機制。flink還用到開辟記憶體池、用二進制資料代替對象、量身定制序列化、定制緩存友好的算法等優化手段。flink還在任務的執行方面進行優化，包括多階段并行執行和增量疊代。
擁抱機器學習和人工智能也是大資料計算的潮流之一。spark和flink分别推出機器學習庫spark ml和flink ml。更多的平台在第三方大資料計算架構上提供機器學習，如mahout、oryx及一幹apache孵化項目systemml、hivemall、predictionio、samoa、madlib。這些機器學習平台一般都同時支援多個計算架構，如mahout同時以spark、flink、h2o為引擎，samoa則使用s4、storm、samza。在深度學習掀起熱潮後，又有社群探索把深度學習架構與現有分布式計算架構結合起來，這樣的項目有sparknet、caffe on spark、tensorframes等。
在同一平台上支援多種架構也是發展趨勢之一，尤其對于那些開發實力較為雄厚的社群。spark以批處理模型為核心，實作了互動式分析架構spark sql、流計算架構spark streaming（及正在實作的structured streaming）、圖計算架構graphx、機器學習庫spark ml。而flink在提供低延遲的流計算的同時，批處理、關系計算、圖計算、機器學習，一個也沒落下，目标直奔大資料通用計算平台。google的beam（意為batch+stream）則試圖把spark、flink、apex這樣的計算架構納入自己制定的标準之下，頗有号令江湖之意。
圖12. beam的統一模型
7. 學習資料
最後介紹一下大資料計算方面的學習資料。入門前的了解、知識面的拓展及知識的零散積累靠長期通路相關的網站、論壇、微信訂閱号，問題解答則靠對搜尋引擎的熟練駕馭。需要指出的是，網上的内容良萎不齊，很多資料是過時的，以訛傳訛也是常有的事，要注意鑒别。
論壇首推知乎、quora、stack overflow，運氣好的話開發者親自給你解答。其他值得關注的網站或論壇包括煉數成金、人大經濟論壇、csdn、部落格園、雲栖社群、360大資料、推酷、伯樂線上、小象學院等。微信訂閱号中，infoq是最權威的，其他還有thu資料派、大資料雜談、csdn大資料、資料猿、hadoop技術博文等，各人根據偏好取舍。
若要進行系統的學習，則首先應參考官方網站文檔。不少大資料平台的官方文檔内容都比較詳實，勝過多數教材。另外，官方文檔與産品通常同步更新，這個優勢是其他資料無法做到的。不過要說可讀性，書籍或視訊教程要強得多。視訊資料可以從上文提到的部分網站論壇下載下傳。
書籍方面，國外o'reilly、manning兩家出版社在大資料領域出版了不少優秀書籍，特别是manning的in action系列和o'reilly的definitive guide系列。前者側重提高動手能力，後者則知識比較全面。in action和definitive guide系列的書籍很多已翻譯為中文，一般分别譯為xxx實戰、xxx權威指南。另外一家出版社packt也值得關注。packt的書比較薄，适合入門。至于中文原創書籍，推薦張俊林的《大資料日知錄》，該書是對大資料存儲和處理技術的全面梳理，系統性強。其他書籍不逐一點評，若想購買或閱讀可參考豆瓣對該書的評分。
圖13. 部分推薦書籍
對希望對大資料架構内部機制有深入的了解的讀者，建議首先檢索相關論文來閱讀。
google的那幾篇論文這裡就不一一列出了，網上很容易搜到。其他推薦的論文如下：
原文釋出時間為：2017-5-10
本文來自雲栖社群合作夥伴“大資料文摘”，了解相關資訊可以關注“bigdatadigest”微信公衆号