相當長一段時間以來，大數據社區已經普遍認識到了批量數據處理的不足。很多應用都對實時查詢和流式處理產生了迫切需求。最近幾年，在這個理念的推動下，催生出了一系列解決方案，Twitter Storm，Yahoo S4，Cloudera Impala，Apache Spark和Apache Tez紛紛加入大數據和NoSQL陣營。本文嘗試探討流式處理系統用到的技術，分析它們與大規模批量處理和OLTP/OLAP數據庫的關系，並探索一個統一的查詢引擎如何才能同時支持流式、批量和OLAP處理。

在Grid Dynamics，我們面臨的需求是構建一個流式數據處理系統，每天需要處理80億事件，並提供容錯能力和嚴格事務性，即不能丟失或重復處理事件。新系統是現有系統的補充和繼任者，現有系統基於Hadoop，數據處理延遲高，而且維護成本太高。此類需求和系統相當通用和典型，所以我們在下文將其描述為規範模型，作為一個抽象問題陳述。

下圖從高層次展示了我們的生產環境概況：

這是一套典型的大數據基礎設施：多個數據中心的各個應用程序都在生產數據，數據通過數據收集子系統輸送到位於中心設施的HDFS上，然後原始數據通過標準的Hadoop工具棧（MapReduce，Pig，Hive）進行匯總和分析，匯總結果存儲在HDFS和NoSQL上，再導出到OLAP數據庫上被定制的用戶應用訪問。我們的目標是給所有的設施配備上新的流式處理引擎（見圖底部），來處理大部分密集數據流，輸送預匯總的數據到HDFS，減少Hadoop中原始數據量和批量job的負載。

流式處理引擎的設計由以下需求驅動：

• SQL-like功能：引擎能執行SQL-like查詢，包括在時間窗口上的join和各種聚合函數，用來實現復雜的業務邏輯。引擎還能處理從匯總數據中加載的相對靜態數據（admixtures）。更復雜的多道次數據挖掘算法不在短期目標範圍內。
• 模塊化和靈活性：引擎絕不僅是簡單執行SQL-like查詢，然後相應的管道就被自動創建和部署，它應該能通過連接各模塊，很方便地組合出更復雜的數據處理鏈。
• 容錯：嚴格容錯是引擎的基本需求。如草圖中，一種可能的設計是使用分布式數據處理管道來實現join、匯總或者這些操作組成的鏈,再通過容錯的持久化buffer來連接這些管道。利用這些buffer，可以實現發布/訂閱的通信方式，可以非常方便地增加或者移除管道，這樣最大程度提升了系統的模塊化。管道也可以是有狀態的，引擎的中間件提供持久化存儲來啟用狀態檢查點機制。所有這些主題將在本文後續章節進行討論。
• 和Hadoop交互：引擎應該能接入流式數據和來自Hadoop的數據，作為HDFS之上的定制化查詢引擎提供服務。
• 高性能和便攜性：即使在最小規模的集群上，系統也能每秒傳遞成千上萬次消息。引擎理應是緊湊高效的，能夠被部署在多個數據中心的小規模集群上。

為了弄明白如何實現這樣一個系統，我們討論以下主題：

• 首先，我們討論流式數據處理系統，批量處理系統和關系型查詢引擎之間的關系，流式處理系統能夠大量用到其他類型系統中已經應用的技術。
• 其次，我們介紹一些在構建流式處理框架和系統中用到的模式和技術。此外，我們調研了當下新興技術，提供一些實現上的小貼士。

The article isbased on a research project developed at Grid Dynamics Labs. Much of the creditgoes to Alexey Kharlamov and Rafael Bagmanov who led the project and othercontributors: Dmitry Suslov, Konstantine Golikov, Evelina Stepanova, AnatolyVinogradov, Roman Belous, and Varvara Strizhkova.

分布式查詢處理基礎

分布式流式數據處理顯然和分布式關系型數據庫有聯系。許多標準查詢處理技術都能應用到流式處理引擎上來，所以理解分布式查詢處理的經典算法，理解它們和流式處理以及其他流行框架比如MapReduce的關系，是非常有用的。

分布式查詢處理已經發展了數十年，是一個很大的知識領域。我們從一些主要技術的簡明概述入手，為下文的討論提供基礎。

分區和Shuffling

分布式和並行查詢處理重度依賴數據分區，將大數據打散分成多片讓各個獨立進程分別進行處理。查詢處理可能包括多步，每一步都有自己的分區策略，所以數據shuffling操作廣泛應用於分布式數據庫中。

盡管用於選擇和投射操作的最優分區需要一定技巧（比如在範圍查詢中），但我們可以假設在流式數據過濾中，使用哈希分區在各處理器中分發數據足以行得通。

分布式join不是那麽簡單，需要深入研究。在分布式環境中，並行join通過數據分區實現，也就是說，數據分布在各個處理器中，每個處理器運行串行join算法（比如嵌套循環join或者排序-合並join或者哈希join）處理部分數據。最終結果從不同處理器獲取合並得來。

分布式join主要采用兩種數據分區技術：

• 不相交數據分區
• 劃分-廣播join

不相交數據分區技術使用join key將數據shuffle到不同分區，分區數據互不重疊。每個處理器在自己分區數據上執行join操作，不同處理器的結果簡單拼接產生最終結果。考慮R和S數據集join的例子，它們以數值鍵k進行join，以簡單的取模函數進行分區。（假設數據基於某種策略，已經分布在各個處理器上）：

下圖演示了劃分-廣播join算法。數據集R被劃分成多個不相交的分區（圖中的R1，R2和R3），數據集S被復制到所有的處理器上。在分布式數據庫中，劃分操作本身通常不包含在查詢過程中，因為數據初始化已經分布在不同節點中。

這種策略適用於大數據集join小數據集，或者兩個小數據集之間join。流式數據處理系統能應用這種技術，比如將靜態數據（admixture）和數據流進行join。

Group By處理過程也依賴shuffling，本質上和MapReduce是類似的。考慮以下場景，數據集根據字符字段分組，再對每組的數值字段求和：

在這個例子中，計算過程包含兩步：本地匯總和全局匯總，這基本和Map/Reduce操作對應。本地匯總是可選的，原始數據可以在全局匯總階段傳輸、shuffle、匯總。

本節的整體觀點是，上面的算法都天然能通過消息傳遞架構模式實現，也就是說，查詢執行引擎可以看做是由消息隊列連接起來的分布式網絡節點組成，概念上和流式處理管道是類似的。

管道

前一節中，我們註意到很多分布式查詢處理算法都類似消息傳遞網絡。但是，這不足以說明流式處理的高效性：查詢中的所有操作應該形成鏈路，數據流平滑通過整個管道，也就是說，任何操作都不能阻塞處理過程，不能等待一大塊輸入而不產生任何輸出，也不用將中間結果寫入硬盤。有一些操作比如排序，天然不兼容這種理念（顯而易見，排序在處理完輸入之前都不能產生任何輸出），但管道算法適用於很多場景。一個典型的管道如下圖所示：

在這個例子中，使用三個處理器哈希join四個數據集：R1，S1，S2和S3。首先並行給S1，S2和S3建立哈希表，然後R1元組逐個流過管道，從S1，S2和S3哈希表中查找相匹配記錄。流式處理天然能使用該技術實現數據流和靜態數據的join。

在關系型數據庫中，join操作還會使用對稱哈希join算法或其他高級變種。對稱哈希join是哈希join算法的泛化形式。正常的哈希join至少需要一個輸入完全可用才能輸出結果（其中一個輸入用來建立哈希表），而對稱哈希join為兩個輸入都維護哈希表，當數據元組抵達後，分別填充：

元組抵達時，先從另外一個數據流對應的哈希表查找，如果找到匹配記錄，則輸出結果，然後元組被插入到自身數據流對應的哈希表。

當然，這種算法對無限流進行完全join不是太有意義。很多場景下，join都作用於有限的時間窗口或者其他類型的緩沖區上，比如用LFU緩存數據流中最常用的元組。對稱哈希join適用於緩沖區大過流的速率，或者緩沖區被應用邏輯頻繁清除，或者緩存回收策略不可預見的場景。在其他情況下，使用簡單哈希join已足夠，因為緩沖區始終是滿的，也不會阻塞處理流程：

值得註意的是，流式處理往往需要采用復雜的流關聯算法，記錄匹配不再是基於字段相等條件，而是基於評分度量，在這種場景下，需要為兩個流維護更為復雜的緩存體系。

流式處理模式

前一節中，我們討論了一些在大規模並行流處理中用到的標準查詢技術。從概念層面上來看，似乎一個高效分布式數據庫查詢引擎能勝任流式處理，反之亦然，一個流式處理系統也應該能充當分布式數據庫查詢引擎的角色。Shuffling和管道是分布式查詢處理的關鍵技術，而且通過消息傳遞網絡能夠自然而然地實現它們。然而真實情況沒那麽簡單。在數據庫查詢引擎中，可靠性不是那麽關鍵，因為一個只讀查詢總是能夠被重新運行，而流式系統則必須重點關註消息的可靠處理。在本節中，我們討論流式系統保證消息傳遞的技術，和其他一些在標準查詢處理中不那麽典型的模式。

流回放

在流式處理系統中，時光倒流和回放數據流的能力至關重要，因為以下原因：

• 這是確保數據正確處理的唯一方式。即使數據處理管道是容錯的，也難以確保數據處理邏輯是無缺陷的。人們總是面臨修復和重新部署系統的需求，需要在新版本的管道上回放數據。
• 問題調查需要即席查詢。如果發生了問題，人們可能需要加入日誌或者修改代碼，在產生問題的數據上重跑系統。
• 即使錯誤不常發生，即使系統總體上是容錯的，流式處理系統也必須設計成在發生錯誤時能夠從數據源中重新讀取特定消息。

因此，輸入數據通常通過緩沖區從數據源流入流式管道，允許客戶端在緩沖區中前後移動讀取指針。

Kafka消息隊列系統就實現了這樣一個緩沖區，支持可擴展分布式部署和容錯，同時提供高性能。

流回放要求系統設計至少考慮以下需求：

• 系統能夠存儲預定義周期內的原始數據。
• 系統能夠撤銷一部分處理結果，回放對應的輸入數據，產出新版本結果。
• 系統能夠快速倒回，回放數據，然後追上源源不斷的數據流進度。

血緣跟蹤

在流式系統中，事件流過一連串的處理器直到終點（比如外部數據庫）。每個輸入事件產生一個由子事件節點（血緣）構成的有向圖，有向圖以最終結果為終點。為了保障數據處理可靠性，整個圖都必須被成功處理，而且在失敗的情況下能重啟處理過程。

實現高效血緣跟蹤是一個難題。我們先介紹Twitter Storm是如何跟蹤消息，保障“至少一次”消息處理語義：

• 數據源（數據處理圖的首個節點）發射的所有事件都標記上一個隨機的eventID。框架為每個數據源的初始事件都維護了一個[eventIDàsignature]的鍵值對集合，其中signature以eventID初始化。
• 下遊節點接收到初始事件後，能產生0個或者多個事件，每個事件攜帶自身的隨機eventID和初始事件的eventID。
• 如果事件被圖中下一個節點成功接收和處理，這個節點會更新初始事件的signature，規則是將初始signature與a）輸入事件的ID和b）產生的所有事件的ID做異或。如圖中第二部分，事件01111產生事件01100,10010和00010，所以事件01111的signature變成11100（=01111（初始值）xor 01111 xor 01100 xor 10010 xor 00010）。
• 一個事件也能基於多個輸入事件生成。這種情況下，事件關聯到多個初始事件，攜帶多個初始ID（圖中第三部分黃色背景事件）。
• 事件的signature變為0代表事件被成功處理。最後一個節點確認圖中的最後一個事件成功處理，而且不再往下遊發送新的事件。框架給事件源節點發送提交消息（如圖第三部分）。
• 框架定期遍歷初始事件表，查找尚未完全處理的事件（即signature不為0的事件）。這些事件被標記為失敗，框架請求源節點回放事件。（譯者註：Storm消息回放不是自動的，可以在消息發送時加上消息ID參數，然後根據失敗的消息ID自行處理回放邏輯，一般Spout對接消息隊列系統，利用消息隊列系統的回放功能。）
• 值得註意的是，由於異或操作的可交換特性，signature的更新順序無關緊要。在下圖中，第二部分中的確認操作可以發生在第三部分之後，這讓完全的異步處理成為可能。
• 還需要註意的是該算法不是嚴格可靠的——一些ID的組合可能偶然導致signature變為0。但是64位長的ID足以保證極低的錯誤概率，出錯概率大概是2^(-64)，在大多數應用中這都是可以接受的。算法的主要優點是只需要少量內存就可以保存下signature表。

以上實現非常優雅，具有去中心化特性：每個節點獨立發送確認消息，不需要一個中心節點來顯式跟蹤血緣。然而，對於維護了滑動窗口或其他類型緩沖區的數據流，實現事務處理變得比較困難。比如，滑動窗口內可能包含成千上萬個事件，很多事件處於未提交或計算中狀態，需要頻繁持久化，管理事件確認過程難度很大。

Apache Spark[3]使用的是另外一種實現方法，其想法是把最終結果看作是輸入數據的處理函數。為了簡化血緣跟蹤，框架分批處理事件，結果也是分批的，每一批都是輸入批次的處理函數。結果可以分批並行計算，如果某個計算失敗，框架只要重跑它就行。考慮以下例子：

在這個例子中，框架在滑動窗口上join兩個流，然後結果再經過一個處理階段。框架把流拆分成批次，每個批次指定ID，框架隨時都能根據ID獲取相應批次。流式處理被拆分為一系列事務，每個事務處理一組輸入批次，使用處理函數轉換數據，並保存結果。在上圖中，紅色加亮部分代表了一次事務。如果事務失敗，框架重跑它，最重要的是，事務是可以並行的。

這種方式簡潔而強大，實現了集中式事務管理，並天然提供“只執行一次”消息處理語義。這種技術還同時適用於批量處理和流式處理，因為不管輸入數據是否是流式的，都把它們拆分成一系列批次。

狀態檢查點

前一節，我們在血緣跟蹤算法中使用簽名（校驗和）提供了“至少一次”消息傳遞語義。該技術改善了系統的可靠性，但留下了至少兩個開放式問題：

• 很多場景都要求“只執行一次”處理語義。比如，如果某些消息被發送兩次，計數器管道會統計出錯誤的結果。
• 處理消息時，管道中的節點計算狀態被更新。計算狀態需要持久化或者復制，以免節點失敗時發生狀態丟失。

Twitter Storm使用以下協議解決這些問題：

• 事件被分組成批次，每個批次關聯一個事務ID。事務ID單調遞增（比如，第一批事件ID為1，第二批ID為2，諸如此類）。如果管道處理某個批次時失敗，這批數據以同樣的事務ID重新發送。
• 首先，框架通知管道中節點新事務啟動。然後，框架發送這一批次數據通過管道。最後，框架通知事務已完成，所有的節點提交狀態變更（比如更新到外部數據庫）。
• 框架保證所有事務的提交是有序的，比如事務2不能先於事務1被提交。這保證了處理節點可以使用以下邏輯持久化狀態變更：
  • 最新的事務ID和狀態被持久化
  • 如果框架請求提交的當前事務ID和數據庫中ID不同，狀態被更新，比如數據庫中的計數器被增加。因為事務的強有序性，所以每個批次數據只會更新一次。
  • 如果當前事務ID和數據庫中ID相同，那麽這是這個批次數據的回放，節點會忽略這次提交。節點應該已經處理過這個批次，並更新過狀態，而事務可能是因為管道中其他部分出錯而失敗。
  • 有序提交對實現“只執行一次”處理語義至關重要。然而，事務嚴格順序處理不太可取，因為下遊所有節點處理完成之前，管道中的首個節點都處於空閑狀態。通過並行處理事務過程，串行化提交能緩解這個問題，如下圖所示：

如果數據源是容錯的，能夠被回放，那麽事務能保障“只執行一次”處理語義。但是，即使使用大容量分批處理，持久化狀態更新也會導致嚴重的性能退化。所以，應該盡可能減少或者避免中間計算結果狀態。補充說明的是，狀態寫入也能通過不同方式實現。最直接的方式是在事務提交過程中，把內存中狀態復制到持久化存儲。這種方式不適用於大規模狀態（比如滑動窗口等）。另一種可選方式是存儲某種事務日誌，比如將原始狀態轉化為新狀態的一系列操作日誌（對滑動窗口來說，可以是一組添加和清理事件）。雖然這種方式需要從日誌中重建狀態，災難恢復變得更麻煩，但在很多場景下，它都能提供更好的性能。

可加性和草圖

中間和最終計算結果的可加性很重要，能極大地簡化流式數據處理系統的設計，實現，維護和恢復。可加性意味著大範圍時間或者大容量數據分區的計算結果能夠由更小的時間範圍或者更小的分區結果組合而來。比如，每日PV量等於每小時PV量之和。狀態可加性允許將數據流切分處理，如我們在前一節討論，每個批次都可以被獨立計算/重算，這有助於簡化血緣跟蹤和減少狀態維護的復雜性。

實現可加性往往不輕松：

• 有一些場景，可加性確實很簡單。比如簡單計數是可加的。
• 在一些場景下，需要存儲一些附加信息來實現可加性。比如，系統統計網店每小時平均購買價格，盡管每日平均購買價格不等於對24個小時平均購買價格再求平均，但是，如果系統還存儲了每個小時的交易量，就能輕易算出每日平均購買價格。
• 在其他場景下，實現可加性非常困難甚至於不可能。比如，系統統計某個網站的獨立訪客數據。假設昨天和今天都有100個獨立用戶訪問了網站，但這兩天的獨立訪問用戶之和可能是100到200之間的任何值。我們不得不維護用戶ID列表，通過ID列表的交集和並集操作來實現可加性。用戶ID列表的大小和處理復雜性和原始數據相當。

草圖(Sketches)是將不可加值轉換為可加值的有效方法。在上面的例子中，ID列表可以被緊湊的可加性統計計數器代替。計數器提供近似值而不是精確值，但在很多應用中都是可以接受的。草圖在互聯網廣告等特定領域非常流行，可以被看做一種獨立的流式處理模式。草圖技術的深入綜述請見[5]。

邏輯時間跟蹤

流式計算中通常會依賴時間：匯總和Join一般作用在滑動時間窗口上；處理邏輯往往依賴事件的時間間隔等。顯然，流式處理系統應該有自己的時間視圖，而不應該使用CPU掛鐘時間。因為發生故障時，數據流和特定事件會被回放，所以實現正確的時間跟蹤並不簡單。通常，全局的邏輯時間概念可以通過以下方式實現：

• 原始系統產生的所有事件都應該標記上時間戳。
• 管道中的處理器處理流時，跟蹤最大時間戳，如果持久化的全局時鐘落後了，就把它更新為最大時間戳。其他處理器與全局時鐘進行時間同步
• 在數據回放時，重置全局時鐘。

持久化存儲匯總

我們已經討論了持久化存儲可以用於狀態檢查點,但這不是流式系統引入外部存儲的唯一作用。考慮使用Cassandra在時間窗口上join多個數據流的場景。不用再維護內存中的事件緩沖區，我們可以把所有數據流的傳入事件保存到Casandra中，使用join key作為row key，如圖所示：

在另一邊，第二個處理器定期遍歷數據記錄，組裝和發送join後的記錄，清理超出時間窗口的事件。Cassandra還可以根據時間戳排序事件來加速處理過程。

不正確的實現會讓整個流式數據處理過程功虧一簣——即使使用Cassandra或者Redis等快速存儲系統，單獨寫入每條數據也會引入嚴重的性能瓶頸。另一方面，使用存儲系統提供了更完善的狀態持久化功能，如果應用批量寫入等優化手段，在很多場景下，也能達成可接受的性能目標。

滑動窗口聚合

流式數據處理經常處理 “過去10分鐘流的某個數據值求和是多少” 等類似查詢，即時間窗口上的連續查詢。針對這類查詢，最直接的解決方案是分別計算各個時間窗口的sum等聚合函數。很顯然，這種方案不是最優的，因為兩個連續的時間窗口實例具有高度相似性。如果時刻T的窗口包含樣本{s(0),s(1),s(2),...,s(T-1),s(T)},那麽時刻T+1的窗口就包含{s(1),s(2),s(3)...,s(T),s(T+1)}。觀察可知可以使用增量處理。

時間窗口之上的增量計算也被廣泛應用在數字信號處理中，包括軟件和硬件。典型例子是計算sum值。如果當前時間窗口的sum值已知，那麽下次時間窗口的sum值就能通過加上新的樣本和減去窗口中最老的樣本得出。

類似技術不僅能用於求和和乘積等簡單聚合函數，也能用於更復雜的轉換過程。比如，SDFT（滑動離散傅裏葉變換）算法[4]就比對每個窗口使用FFT（快速傅裏葉變換）算法要高效得多。

查詢處理管道：Storm, Cassandra, Kafka

現在回到文章一開始提出的實際問題上來。我們基於Storm，Kafka和Cassandra（這些組件應用了前文介紹的技術）設計和實現了自己的流式處理系統。在此，我們僅提供解決方案的簡明概述——詳細描述所有實現上的坑和技巧的篇幅太長，可能需要單獨一篇文章。

系統理所當然使用Kafka0.8。Kafka作為分區、容錯的事件緩沖區，能夠實現流回放，可以輕松添加新的事件生產者和消費者，增強了系統的擴展性。Kafka讀指針回溯的能力也使隨機訪問傳入的數據批次成為可能，相應地，可以實現Spark風格的血緣跟蹤，也可以將系統輸入指向HDFS處理歷史數據。

如之前描述，Cassandra用於實現狀態檢查點和持久化存儲聚合。在很多使用場景中，Cassandra也用於存儲最終結果。

TwitterStorm是系統的基石。所有的活動查詢處理都運行於Storm的topologies中，topologies和Kafka、Cassandra進行交互。一些數據流是簡單的：數據抵達Kafka；Storm讀取並處理，然後把結果存儲在Cassandra或者其他地方。其他數據流更為復雜：一個Storm topology通過Kafka或Cassandra將數據傳遞給另一個topology。上圖展示了兩個此類型數據流（紅色和藍色曲線箭頭）。

邁向統一大數據處理平臺

現有的Hive，Storm和Impala等技術讓我們處理大數據時遊刃有余，復雜分析和機器學習時使用批量處理，在線分析時使用實時查詢處理，連續查詢時使用流式處理。更進一步，Lambda架構還能有效整合這些解決方案。這給我們帶來的問題是：將來這些技術和方法怎樣才能聚集成一個統一解決方案。本節我們討論分布式關系查詢處理，批量處理和流式處理最突出的共同點，合計出能夠覆蓋所有用戶場景，從而在這個領域最具發展潛力的解決方案。

關鍵之處在於，關系型查詢處理，MapReduce和流式處理都能通過shuffling和管道等相同的概念和技術來實現。同時：

• 流式處理必須保障嚴格的數據傳遞和中間狀態持久化。這些特性對於計算過程很方便重啟的批量處理不太關鍵。
• 流式處理離不開管道。對於批量處理，管道不那麽關鍵甚至在有些場景下不適用。Apache Hive就是基於分階段的MapReduce過程，中間結果被物化，沒有完全利用上管道的優點。

以上兩點暗示，可調的持久化策略（基於內存的消息傳遞或者存儲在硬盤上）和可靠性是我們想象中統一查詢引擎的顯著特性。統一查詢引擎為高層的框架提供一組處理原語和接口。

在新興技術中，以下兩者值得重點關註：

• Apache Tez[8],Stinger Initiative[9]的一部分。Apache Tez通過引入一組細粒度的查詢處理原語來替代MapReduce框架。它的目標是讓Apache Pig和Apache Hive等框架把查詢語句和腳本分解成高效的查詢處理管道，而不是一系列的MapReduce Job，後者通常很慢，因為需要存儲中間結果。
• Apache Spark[10].Spark項目可能是最先進和最有前途的統一大數據處理平臺，它已經包含了批量處理框架，SQL查詢引擎和流式處理框架。

References

1. Wilschut and P. Apers,“Dataflow Query Execution in a Parallel Main-Memory Environment “
2. T. Urhan and M. Franklin,“XJoin: A Reactively-Scheduled Pipelined Join Operator“
3. M. Zaharia, T. Das, H. Li, S.Shenker, and I. Stoica, “Discretized Streams: An Ef?cient and Fault-TolerantModel for Stream Processing on Large Clusters”
4. E. Jacobsen and R. Lyons, “The Sliding DFT“
5. Elmagarmid, Data Streams Modelsand Algorithms
6. N. Marz, “Big Data Lambda Architecture”
7. J. Kinley, “The Lambda architecture: principles for architecting realtimeBig Data systems”
8. http://hortonworks.com/hadoop/tez/
9. http://hortonworks.com/stinger/
10. http://spark-project.org/

翻譯-In-Stream Big Data Processing 流式大數據處理