一個實時流處理框架通常需要兩個基礎(chǔ)架構(gòu)：處理器和隊列。處理器從隊列中讀取事件，執(zhí)行用戶的處理代碼，如果要繼續(xù)對結(jié)果進行處理，處理器還會把事件寫到另外一個隊列。隊列由框架提供并管理。隊列做為處理器之間的緩沖，傳輸數(shù)據(jù)和事件，這樣處理器可以單獨操作和擴展。例如，一個web 服務訪問日志處理應用，可能是這樣的：

框架之間的主要區(qū)別在于隊列語義，通常不同之處有以下幾點：

• 調(diào)度保障機制: 至少一次，至多一次，只有一次。

• 容災機制: 失敗對用戶和自動恢復是透明的。

• 可用性: 數(shù)據(jù)在出現(xiàn)錯誤后可以保存并重啟。

• 可擴展性:產(chǎn)品/用戶增加時的局限性。

• 性能Performance: 隊列操作的吞吐量和延遲。

我們想在開源的 Cask Data Application Platform (CDAP)上提供一個動態(tài)可擴展，強一致性并且有一次性交易機制的實時流處理框架，在這個強大的機制保護下，開發(fā)者可以自由操作任何形式的數(shù)據(jù)操作而不用擔心不一致性，潛在的返工和失敗。它可以幫助開發(fā)者在沒有分布式系統(tǒng)背景的情況下建立他們的大數(shù)據(jù)應用。此外，如果需要可以關(guān)閉這種強大的保護機制換取高性能。它總是比其他方式更容易使用。

可擴展隊列

隊列有兩種基本操作：入隊和出隊。生產(chǎn)者將消息寫到隊頭（入隊），消費者從隊尾讀取數(shù)據(jù)（出隊）。如果做為一個整體你添加更多生成者時的入隊速度和添加更多消費者時的出隊速度足夠快，我們說這個隊列是可擴展的。

理想狀態(tài)下，擴展是線性的，這意味著兩倍的生產(chǎn)者 /消費者，會產(chǎn)生兩位速度的出隊/入隊，增長只受集群的規(guī)模限制。為了支持生產(chǎn)者的線性擴展，隊列需要一個存儲系統(tǒng)并且需要當前寫入者的數(shù)量線性擴展。為了應對消費者的線性擴展，隊列可以分區(qū)，例如一個消費者只處理隊列中的一段數(shù)據(jù)。

隊列擴展的另一個方面是它應該可以橫向擴展。這意味著隊列性能的上限可以通過增加集群結(jié)點的方式來提升。這是很重要的，它可以保證隊列不受當前集群大小限制根據(jù)數(shù)據(jù)的增長而擴展。

分區(qū)的 HBase 隊列

我們選擇 Apache HBase做為隊列的存儲層。它為存儲強一致性，可橫向擴展的行數(shù)據(jù)做了設計和優(yōu)化。它的并發(fā)寫操作性能非常好，并提供了有序掃描以支持分區(qū)消費者。我們使用 HBase Coprocessors 的高效掃描濾波和隊列清洗。為了在隊列上使用一次性語義，我們用 Tephra’s 為 HBase 提供傳輸支持。

生產(chǎn)者和消費者具有操作獨立性。每個生產(chǎn)者通過 Hbase puts 批處理執(zhí)行入隊操作，消費者通過執(zhí)行 Hbase Scans 執(zhí)行出隊操作。生產(chǎn)者和消費者的數(shù)量之間沒有關(guān)聯(lián)，他們可以分離。

此隊列存在一個消費者組的概念。一個消費者組，是由相同的關(guān)鍵字劃分的消費者集合，這樣，每個發(fā)布到隊列的事件，就會由此消費者組中的消費者去消費。使用消費者組，可以通過不同的關(guān)鍵字劃分同一個隊列，同時，也可以通過數(shù)據(jù)的操作性特點來拓展。按照上面訪問日志分析的例子，生產(chǎn)者和消費者組可能看起來像這樣：

隊列行值格式

當一個事件通過一個生產(chǎn)者被發(fā)布出去，一個或多個消費者組合將收到消息，我們把事件寫入 HBase 表的一個或多個行上，那么這條記錄就被設計成適用于每個消費者組。事件的有效負荷和元數(shù)據(jù)被存儲在獨立的列上，那么行的值就是下面這樣的格式：

兩個有趣的部分是行的值是分區(qū) ID 和整個 ID。分區(qū) ID 通過限定行值前綴再提供給消費者。消費者只被允許讀數(shù)據(jù)，并在出隊的時候使用前綴掃描。分區(qū) ID 由兩部分組成：一個消費者組 ID 和一個消費者 ID。生產(chǎn)者計算出每個消費者組的分區(qū) ID，并通過入隊寫到那些行。

行關(guān)鍵字中的入口 ID（Entry ID）包含了事務信息。它由 Tephra 觸發(fā)的生產(chǎn)者事務寫指針和單向增長的計數(shù)器組成。這個計數(shù)器由本地的生產(chǎn)者生成，同時，針對事件，計數(shù)器需要讓行關(guān)鍵字唯一，因為生產(chǎn)者可以在同一個事務中將多個事件加入隊列。

出隊列的時候，計數(shù)器會使用事務寫指針來決定，隊列入口是否已經(jīng)提交，以及是否可以消費了。事務寫指針和計數(shù)器的組合，使得行關(guān)鍵字總是唯一的。這讓生產(chǎn)者可以獨立的操作，而不會有寫沖突。

為了生成分區(qū) ID（Partition ID），生產(chǎn)者需要知道大小和每個消費者組的分區(qū)關(guān)鍵字。當應用程序啟動，以及組大小發(fā)生任何變化的時候，消費者組信息都會被記錄下來。

改變生產(chǎn)者和消費者

增加或減少生產(chǎn)者是很直接的，因為每個生產(chǎn)者都是獨立操作的。增加或減少生產(chǎn)者進程就可以滿足這個要求。然而，當消費者組的大小需要改變的時候，就需要協(xié)調(diào)來正確更新消費者組的信息。可以用下面的圖表來概括所需的步驟：

由于暫停和恢復是由 Apache ZooKeeper 來協(xié)調(diào)的，同時它們也是并行執(zhí)行的，所以它們是兩個非?？焖俚牟僮?。例如，之前我們提到的 Web 訪問日志分析應用程序，改變消費者組信息的過程可能看起來像這樣：

基于這個隊列的設計，入隊列和出隊列的性能，與單獨的批量 HBase Puts 和 HBase Scans 不相上下，這樣也帶來與 Tephra 服務器進行通訊的開銷。通過在同一個業(yè)務處理中將多個事件批量處理，可以大大降低這個開銷。

最后，為了避免“熱點聚焦（hotspotting）“，我們基于簇的大小提前分割了 HBase 表，同時，在行關(guān)鍵字（row key）上采用 加鹽（salting） 的方式來更好的分配寫。否則，由于是單調(diào)的增加業(yè)務處理寫指針，行關(guān)鍵字就會是連續(xù)的。

性能值

我們在小型的 10 節(jié)點的 HBase 集群上已經(jīng)測試過性能，結(jié)果令人印象深刻。使用 1K 字節(jié)負載，以 500 個事件為一個批次大小，我們完成了生產(chǎn)和消費 100K 個事件/秒的吞吐量，其中運行了 3 個生產(chǎn)者和 10 個消費者。我們也觀察到當我們增加消費者和消費者的時候，吞吐量線性增加：例如，當我們將生產(chǎn)者和消費者數(shù)量加倍的時候，吞吐量增加到 200K 個事件/秒。

在 HBase 的幫助下，結(jié)合最佳實踐，我們成功的創(chuàng)建了一個線性可伸縮的，分布式事務隊列系統(tǒng)。同時，在 CDAP 中使用這個系統(tǒng)提供實時流處理框架：動態(tài)可伸縮，強一致性，以及一次交付的傳輸保證。