帶有Apache Spark的Lambda架構

歡迎你們前往騰訊雲+社區，獲取更多騰訊海量技術實踐乾貨哦~

目標

市場上的許多玩家已經創建了成功的MapReduce工做流程來天天處理以TB計的歷史數據。可是誰願意等待24小時才能得到最新的分析結果？這篇博文將向您介紹旨在利用批處理和流處理方法的Lambda架構。咱們將利用Apache Spark（Core，SQL，Streaming），Apache Parquet，Twitter Stream等實時流數據快速訪問歷史數據。還包括清晰的代碼和直觀的演示！

簡史

Apache Hadoop的豐富歷史始於2002年。Hadoop由Doug Cutting建立，Doug Cutting是Apache Lucene（一個被普遍使用的文本搜索庫）的建立者。Hadoop起源於Apache Nutch，一個開源的網絡搜索引擎，它自己就是Lucene項目的一部分。它在10年前成爲一個獨立的項目。

所以，大量客戶實施了有效的基於Hadoop的M/R處理管道。現實生活中有一些很好的例子：

• Oozie編排的工做流程天天運行並處理高達150 TB的數據以生成分析結果
• bash管理的工做流程天天運行並處理高達8 TB的數據以生成分析結果

現狀

商業現實已經發生了變化，因此如今更快作出的決定更有價值。除此以外，技術也在不斷髮展。Kafka，Storm，Trident，Samza，Spark，Flink，Parquet，Avro，Cloud providers等都是工程師和企業普遍採用的流行語。

所以，現代基於Hadoop的M/R管道（使用Kafka，Avro和數據倉庫等現代二進制格式，即Amazon Redshift，用於臨時查詢）可能採用如下方式：

這看起來至關不錯，但它仍然是一種傳統的批處理方式，具備全部已知的缺點，主要緣由是客戶端的數據在批處理花費大量時間完成以前的數據處理時，新的數據已經進入而致使數據過期。

Lambda架構

Nathan Marz針對通用的，可擴展的和容錯的數據處理架構提出了術語Lambda Architecture。它是一種旨在經過利用批處理和流處理這二者的優點來處理大量數據的數據處理架構。

我強烈建議閱讀Nathan Marz的書，由於它從提出者的角度提供了Lambda Architecture的完整表述。

圖層

從宏觀角度看，它的處理流程以下：

全部進入系統的數據都被分配到批處理層和速度層進行處理。批處理層管理主數據集（一個不可變的，僅可擴展的原始數據集）並預先計算批處理視圖。服務層對批處理視圖進行索引，以即可以在低延遲的狀況下進行點對點查詢。速度層只處理最近的數據。任何傳入的查詢都必須經過合併來自批量視圖和實時視圖的結果來獲得結果。

焦點

許多工程師認爲Lambda Architecture是所有關於這些層次和定義的數據流的，但Nathan Marz在他的書中將重點放在其餘重要方面，如：

• 思考的分佈式
• 避免增量架構
• 強制數據不可變
• 建立從新計算算法

數據的相關性

如前所述，任何傳入查詢都必須經過合併來自批量視圖和實時視圖的結果來獲得答案，所以這些視圖須要可合併性。須要注意的一點是，實時視圖是之前的實時視圖和新數據增量的函數，所以可使用增量算法。批處理視圖是全部數據的函數，所以應該在那裏使用重算算法。

權衡

咱們生活中的每一件事都是一種折衷，而Lambda Architecture也不是一個例外。一般，咱們須要解決一些主要的折衷：

• 徹底從新計算與部分從新計算
  • 在某些狀況下，可使用Bloom過濾器來避免徹底從新計算
• 重算算法與增量算法
  • 使用增量算法有很大的誘惑力，但根據指南咱們必須使用從新計算算法，即便它使達到相同的結果變得更加困難。
• 加法算法與近似算法
  • Lambda Architecture與加法算法很好地協做。所以，這是咱們須要考慮使用近似算法的另外一種狀況，例如，HyperLogLog用於計數不一樣的問題等。

實現

有多種實現Lambda體系結構的方法，由於它對於每一個層的底層解決方案都是不可知的。每一層都須要底層實現的特定功能，這可能有助於作出更好的選擇並避免過分的決定：

• 批處理層：一次寫入，批量讀取屢次
• 服務層：隨機讀取，不隨機寫入; 批量計算和批量寫入
• 速度層：隨機讀取，隨機寫入; 增量計算

例如，其中一個實現（使用Kafka，Apache Hadoop，Voldemort，Twitter Storm，Cassandra）可能以下所示：

Apache Spark

Apache Spark能夠被視爲在全部Lambda體系結構層上處理的集成解決方案。它包含Spark Core，包括高層次的API，而且支持通用執行圖表的優化引擎，Spark SQL爲SQL和結構化數據提供處理，以及Spark Streaming，支持可擴展性，高吞吐量，容錯流的實時數據流的處理。固然，使用Spark進行批量處理可能會很是昂貴，而且可能不適合全部場景和數據量，但除此以外，它是Lambda Architecture實施方案的適當匹配。

示例應用程序

讓咱們用一些捷徑建立一個示例應用程序來演示Lambda架構，這個程序的主要目標是提供在#morningatlohika推文中使用的主題標籤統計數據。

源代碼位於GitHub上，關於上述主題的更多視覺信息位於Slideshare上。

批處理視圖

爲了簡單起見，假設咱們的主數據集包含自開始以來的全部推文。另外，咱們實施了批量處理，建立咱們業務目標所需的批處理視圖，所以咱們有一個預先計算的批處理視圖，其中包含與#morningatlohika一塊兒使用的全部主題標籤統計信息：

apache – 6 
architecture – 12 
aws – 3 
java – 4 
jeeconf – 7 
lambda – 6 
morningatlohika – 15 
simpleworkflow – 14 
spark – 5複製代碼

數字很容易記住，由於我簡單地在相應的主題標籤中使用了許多字母。

實時視圖

想象一下，當應用程序啓動並運行時，如今有人正在發送推文消息：

「@tmatyashovsky關於 #lambda #architecture使用 #apache #spark在 #morningatlohika的酷博客文章 」

在這種狀況下，適當的實時視圖應該包含如下hash標籤和它們的統計信息（在咱們的例子中僅爲1，由於相應的hash標籤只用了一次）：

apache – 1 
architecture – 1 
lambda – 1 
morningatlohika – 1 
spark – 1複製代碼

查詢

當客戶端爲了實時獲得全部的Hash標籤的統計結果進行查詢時，咱們只須要將批量視圖與實時視圖合併便可。因此輸出應該以下所示（適當的hashtags的統計數字增長1）：

apache – 7 
architecture – 13 
aws – 3 
java – 4 
jeeconf – 7 
lambda – 7 
morningatlohika – 16 
simpleworkflow – 14 
spark – 6複製代碼

演示方案

演示場景的簡化步驟以下：

• 經過Apache Spark 建立批處理視圖（.parquet）
• 在Apache Spark中緩存批處理視圖
• 開始鏈接到Twitter的流應用程序
• 關注即時#morningatlohika推文
• 構建增量的實時視圖
• 查詢，即即時合併批處理和實時視圖

技術細節

源代碼基於Apache Spark 1.6.x，即在引入結構化流式傳輸以前。Spark Streaming架構是純粹的微批處理架構：

所以，對於流媒體應用程序，我是用DSTREAM使用鏈接到Twitter TwitterUtils：

JavaDStream < Status > twitterStatuses = TwitterUtils 。createStream （ javaStreamingContext ，
                                                                createTwitterAuthorization （）， 
                                                                new  String [ ] { twitterFilterText } ）;複製代碼

在每一個微批處理中（使用可配置的批處理間隔），我正在執行新推文中hashtags統計的計算，並使用updateStateByKey（）有狀態轉換更新實時視圖的狀態。爲了簡單起見，使用臨時表將實時視圖存儲在內存中。

查詢服務反映了經過代碼顯式合併由DataFrame表示的批處理視圖和實時視圖：

DataFrame realTimeView = streamingService . getRealTimeView ( ) ; 
DataFrame batchView = servingService . getBatchView ( ) ; 
DataFrame mergedView = realTimeView . unionAll ( batchView ) 
                                   . groupBy ( realTimeView . col ( HASH_TAG . getValue ( ) ) ) 
                                   . sum ( COUNT . getValue ( ) )
                                   . orderBy ( HASH_TAG . getValue ( ) ) ;

List < Row > merged = mergedView . collectAsList ( ) ;

return merged . stream ( ) 
   . map ( row - >  new  HashTagCount ( row . getString ( 0 ) , row . getLong ( 1 ) ) ) 
   . collect ( Collectors . toList ( ) ) ;複製代碼

結果

使用簡化的方法，開頭提到的真正基於Hadoop的M/R管道可能會使用Apache Spark進行加強，並按如下方式查看：

並非後記

正如前面提到的，Lambda Architecture有其優勢和缺點，人們也劃分紅支持者和反對者兩派。他們中的一些人說批處理視圖和實時視圖有不少重複的邏輯，由於他們最終須要從查詢角度建立可合併的視圖。因此他們建立了Kappa架構 - 簡化了Lambda架構。Kappa架構系統是刪除了批處理系統的架構。要取代批處理，數據只需經過流式傳輸系統快速提供：

但即便在這種狀況下，Kappa Architecture也有使用Apache Spark的地方，例如流處理系統：

問答
如何使用MySQL和ApacheSPark？
相關閱讀
大數據系統的Lambda架構
Spark生態頂級項目彙總
大數據平臺架構技術選型與場景運用

此文已由做者受權騰訊雲+社區發佈，原文連接：https://cloud.tencent.com/developer/article/1138267?fromSource=waitui