【轉】MapReduce Hold不住？

本文系統地介紹和分析比較了業界主流的Yahoo! S四、StreamBase和Borealis三種流式計算系統，但願讀者能從這些系統的設計中領悟到不一樣場景下流式計算所要解決的關鍵問題。

背景

非實時計算幾乎都基於MapReduce計算框架，但MapReduce並非萬能的。對於搜索應用環境中的某些現實問題，MapReduce並不能很好地解決問題。

商用搜索引擎，像Google、Bing和Yahoo!等，一般在用戶查詢響應中提供結構化的Web結果，同時也插入基於流量的點擊付費模式的文本 廣告。爲了在頁面上最佳位置展示最相關的廣告，經過一些算法來動態估算給定上下文中一個廣告被點擊的可能性。上下文可能包括用戶偏好、地理位置、歷史查 詢、歷史點擊等信息。一個主搜索引擎可能每秒鐘處理成千上萬次查詢，每一個頁面均可能會包含多個廣告。爲了及時處理用戶反饋，須要一個低延遲、可擴展、高可 靠的處理引擎。然而，對於這些實時性要求很高的應用，儘管MapReduce做了實時性改進，但仍很難穩定地知足應用需求。由於Hadoop爲批處理做了 高度優化，MapReduce系統典型地經過調度批量任務來操做靜態數據；而流式計算的典型範式之一是不肯定數據速率的事件流流入系統，系統處理能力必須 與事件流量匹配，或者經過近似算法等方法優雅降級，一般稱爲負載分流（load-shedding）。固然，除了負載分流，流式計算的容錯處理等機制也和 批處理計算不盡相同。

最近Facebook在Sigmod 11上發表了利用HBase/Hadoop進行實時數據處理的論文，經過一些實時性改造，讓批處理計算平臺也具有實時計算的能力。這類基於MapReduce進行流式處理的方案有三個主要缺點。

• 將輸入數據分隔成固定大小的片斷，再由MapReduce平臺處理，缺點在於處理延遲與數據片斷的長度、初始化處理任務的開銷成正比。小的分段會 下降延遲，增長附加開銷，而且分段之間的依賴管理更加複雜（例如一個分段可能會須要前一個分段的信息）；反之，大的分段會增長延遲。最優的分段大小取決於 具體應用。

• 爲了支持流式處理，MapReduce須要被改形成Pipeline的模式，而不是Reduce直接輸出；考慮到效率，中間結果最好只保存在內存中等。這些改動使得原有的MapReduce框架的複雜度大大增長，不利於系統的維護和擴展。

• 用戶被迫使用MapReduce的接口來定義流式做業，這使得用戶程序的可伸縮性下降。

綜上所述，流式處理的模式決定了要和批處理使用很是不一樣的架構，試圖搭建一個既適合流式計算又適合批處理計算的通用平臺，結果可能會是一個高度複雜的系統，而且最終系統可能對兩種計算都不理想。

目前流式計算是業界研究的一個熱點，最近Twitter、LinkedIn等公司相繼開源了流式計算系統Storm、Kafka等，加上 Yahoo!以前開源的S4，流式計算研究在互聯網領域持續升溫。不過流式計算並不是最近幾年纔開始研究，傳統行業像金融領域等很早就已經在使用流式計算系 統，比較知名的有StreamBase、Borealis等。

本文簡單介紹幾種業界使用的流式計算系統，但願流式系統的設計者或開發者們能從中得到啓示。

圖1 數據分析系統總體組成示意圖

圖1從整個分析系統的架構角度，給出了實時計算子系統所處的位置。實時計算系統和批處理計算系統同屬於計算這個大的範疇，批處理計算能夠是 MapReduce、MPI、SCOPE等，實時計算能夠是S四、Storm等，批處理和實時均可以或不依賴統一的資源調度系統。另外，計算系統的輸入、 輸出，包括中間過程的輸入、輸出，都與存儲系統交互，能夠是塊存儲系統HDFS，也能夠是K-V存儲系統Hypertable等。計算層的上層是數據倉 庫，或者直接和用戶交互，交互方式能夠是SQL-like或者MR-like等。

系統

S4

S4是一個通用的、分佈式的、可擴展的、分區容錯的、可插拔的流式系統。基於S4框架，開發者能夠輕鬆開發面向持續流數據處理的應用。

S4的設計特色有如下幾個方面。

• Actor Model

爲了能在普通機型構成的集羣上進行分佈式處理，而且集羣內部不使用共享內存，S4架構採用了Actor模式，這種模式提供了封裝和地址透明語義，因 此在容許應用大規模併發的同時，也提供了簡單的編程接口。S4系統經過處理單元（Processing Elements，PEs）進行計算，消息在處理單元間以數據事件的形式傳送，PE消費事件，發出一個或多個可能被其餘PE處理的事件，或者直接發佈結 果。每一個PE的狀態對於其餘PE不可見，PE之間惟一的交互模式就是發出事件和消費事件。框架提供了路由事件到合適的PE和建立新PE實例的功能。S4的 設計模式符合封裝和地址透明的特性。

• Decentralized and Symmetric Architecture

除了遵循Actor模式，S4也參照了MapReduce模式。爲了簡化部署和運維，從而達到更好地穩定性和擴展性，S4採用了對等架構，集羣中的 全部處理節點都是等同的，沒有中心控制。這種架構將使得集羣的擴展性很好，處理節點的總數理論上無上限；同時，S4將沒有單點容錯的問題。
Pluggable Architecture
S4系統使用Java開發，採用了極富層次的模塊化編程，每一個通用功能點都儘可能抽象出來做爲通用模塊，並且儘量讓各模塊實現可定製化。

• Partial Fault-Tolerance

基於Zookeeper服務的集羣管理層將會自動路由事件從失效節點到其餘節點。除非顯式保存到持久性存儲，不然節點故障時，節點上處理事件的狀態會丟失。

• Object Oriented

節點間通訊採用「Plain Old Java Objects」（POJOs）模式，應用開發者不須要寫Schemas 或用哈希表來在節點間發送Tuples。

S4的功能組件分3大類，Clients、Adapters和PNode Cluster，圖2顯示了S4系統框架。

 

圖2 Yahoo! S4流式系統框架結構圖

S4提供Client Adapter，容許第三方客戶端向S4集羣發送事件和接收事件。Adapter實現了基於JSON的API，支持多語言實現的客戶端驅動。

Client經過Driver組件與Adapter進行交互，Adapter也是一個Cluster，其中有多個Adapter結點，Client 能夠經過多個Driver與多個Adapter進行通訊，這樣能夠保證單個Client在分發大數據量時Adapter不會成爲瓶頸，也能夠確保系統支持 多個Client應用併發執行的快速、高效和可靠性。

在Adapter中，真正與Client交互的是其Stub組件，該組件實現了管理Client與Adapter之間經過TCP/IP協議進行通訊 的功能。GenericJsonClientStub這個類支持將事件在Client與Adapter之間以JSON的形式轉換，從而支持更多種類型的 Client應用。不一樣的Client能夠配置不一樣的Stub來與Adapter進行通訊，用戶能夠定義本身的Stub來實現本身想要的業務邏輯，這樣也 使得Client的行爲更加多樣性、個性化。

StreamBase

StreamBase是IBM開發的一款商業流式計算系統，在金融行業和政府部門使用，其自己是商業應用軟件，但提供了Develop Edition。相對於付費使用的Enterprise Edition，前者的功能更少，但這並不妨礙咱們從外部使用和API接口來對StreamBase自己進行分析。

StreamBase使用Java開發，IDE是基於Eclipse進行二次開發，功能很是強大。StreamBase也提供了至關多的 Operator、Functor以及其餘組件來幫助構建應用程序。用戶只須要經過IDE拖拉控件，而後關聯一下，設置好傳輸的Schema而且設置一下 控件計算過程，就能夠編譯出一個高效處理的流式應用程序了。同時，StreamBase還提供了類SQL語言來描述計算過程。

StreamBase的組件交互狀況如圖3所示。

 

圖3 StreamBase組件交互圖

StreamBase Server是節點上啓動的管理進程，它負責管理節點上Container的實例，每一個Container經過Adapter得到輸入，交給應用邏輯進行計算，而後經過Adapter進行輸出。各個Container相互鏈接，造成一個計算流圖。

Adapter負責與異構輸入或輸出交互，源或目的地可能包括CSV文件、JDBC、JMS、Simulation（StreamBase提供的流產生模擬器）或用戶定製。
每一個StreamBase Server上面都會存在一個Sytsem Container，主要是產生系統監控信息的流式數據。

HA Container用於容錯恢復，能夠看出它實際包含兩個部分：Heartbeat和HA Events，其中HeartBeat也是Tuple在Container之間傳輸。在HA方案下，HA Container監控Primary Server的活動狀況，而後將這些信息轉換成爲HA Events交給StreamBase Monitor來處理。

Monitor就是從System Container和HA Container中獲取數據而且進行處理。StreamBase認爲HA 問題應該經過CEP方式處理，也就是說若是哪一個部件出現問題，就確定會反映在System Container和HA Container的輸出流上面，而後 Monitor經過復瑣事件處理這些Tuples的話就可以檢測到機器故障等問題，並做出相應處理。

StreamBase提出瞭如下4種模板策略來解決容錯問題。

• Hot-Hot Server Pair Template

Primary Server和Secondary Server都在同時計算，而且將計算結果交給下游。優勢是Primary Server若是故障的話那麼Secondary Server依然工做，幾乎沒有任何切換時間；而且下游只須要選取先到來的Tuple就能夠處理了，保證處理速度最快；缺點是浪費計算和網絡資源。

• Hot-Warm Server Pair Template

Primary Server和Secondary Server都在同時計算，但只有Primary Server將計算結果交給下游。優勢是若是Primary Server故障，Secondary Server能夠很快切換，而不須要任何恢復狀態的工做。相對於Hot-Hot方式時間稍微長一些，但沒有Hot-Hot那麼耗費網絡資源，同時也浪費了 計算資源。

• Shared Disk Template

Primary Server在計算以後，將計算的一些中間關鍵狀態存儲到磁盤、SAN（Storage Area Network）或是可靠的存儲介質。若是Srimary Server故障，Secondary Server會從介質中讀取出關鍵狀態，而後接着繼續計算。優勢是沒有浪費任何計算和網路資源，但恢復時間依賴狀態的量級而定，相對於前兩種，恢復時間可 能會稍長。

• Fast Restart Template

這種方案限定了應用場景，只針對無狀態的應用。對於無狀態的狀況，方案能夠很是簡單，只要發現Primary Server故障，Secondary Server當即啓動，並接着上游的數據流繼續計算便可。

Borealis

Borealis是Brandeis University、Brown University和MIT合做開發的一個分佈式流式系統，由以前的流式系統Aurora、Medusa演化而來。目前Borealis系統已經中止維 護，最新的Release版本中止在2008年。

Borealis具備豐富的論文、完整的用戶/開發者文檔，系統是C++實現的，運行於x86-based Linux平臺。系統是開源的，同時使用了較多的第三方開源組件，包括用於查詢語言翻譯的ANTLR、C++的網絡編程框架庫NMSTL等。

Borealis系統的流式模型和其餘流式系統基本一致：接受多元的數據流和輸出，爲了容錯，採用肯定性計算，對於容錯性要求高的系統，會對輸入流使用算子進行定序。

Borealis的系統架構如圖4所示。

• Query Processor（QP）是計算執行的地方，是系統的核心部件，其大部分功能繼承自Aurora。

• I/O Queues將數據流導入QP，路由Tuples到其餘節點或客戶端程序。

• Admin模塊用來控制本地的QP，例如創建查詢、遷移數據流圖片斷，該模塊也會同Local Optimizer協做優化現有數據流圖。

• Local Optimizer職責包括本地調度策略、調整Operator行爲、超載後丟棄低價值元組等。

• Storage Manager模塊用於存儲本地計算的狀態數據。

• Local Catalog存儲本地數據流圖和元數據，能夠被本地全部組件訪問。

• Borealis Node還有彼此通訊的模塊用於執行協做任務。

• Neighborhood Optimizer使用本地和鄰居節點來優化節點間的負載均衡或shed load。

• High Availability （HA）模塊相互監測，發現對方故障時及時代替對方。

• Local Monitor收集本地性能相關統計數字報告給本地和Neighborhood Optimizer。

• Global Catalog爲整個數據流計算提供了一個邏輯上的完整視圖。

除做爲基本功能節點外，Borealis Server也能夠被設計成一個協做節點來執行全局的系統監控和其餘優化任務，好比全局的負載分佈和Global Load Shedding，所以Borealis實際上提供了完整的3級監控和優化（Local、Neighborhood、Global）。

負載均衡方面，Borealis提供了動態和靜態兩種部署機制。

• Correlation-based Operator Distribution

經過分析不一樣Operators和Nodes間的負載變化的關係，決定和動態調整Operatpr的部署，使之達到負載均衡。

• Resilient Operator Distribution Algorithm

該算法的目標是提供一種靜態的Operator部署方案，該方案可以在不須要從新調整的狀況下處理最大可能的輸入速度變化範圍。

因爲動態調整須要時間和消耗，前者適用於負載變化持續時間較長的系統；然後者則能處理較快較短的負載峯值。在實現上前者使用相關係數做爲節點關聯度 指標，並經過貪婪算法將NP問題轉化爲多項式求解；然後者在部署前計算完畢，保證系統可以容忍負載峯值。該算法在線性代數上建模，包括Operator Ordering、Operator Assignment兩個階段。

Borealis經過四種容錯機制來知足用戶需求。

• Amnesia Backup

備機發現主機故障，當即從一個空的狀態開始重作。

• Passive Standby

主機處理，備機待命，主機按週期作Checkpoint，主機故障後切換到備機，重放Checkpoint和數據流，對於不肯定性計算能夠很好地支持，缺點是恢復時間較長。

• Active Standby

主備機同時計算，主機故障時直接切換到備機，不支持不肯定性計算，浪費計算資源，不過恢復時間幾乎沒有。

• Upstream Backup

經過上游備份來容錯，故障時從上游重放數據便可，恢復時間最長，不過最節省資源。

除此以外，Borealis還提供了更高級的容錯機制Rollback Recovery，它是一種基於副本在節點失效、網絡失效或網絡分區時的故障恢復機制，在儘可能減小系統不一致的狀況下，儘量地保證系統的可用性。該機制 容許用戶定義一個閾值來在一致性和可用性之間作一個平衡。當系統數據恢復後，系統支持從新計算輸出正確的結果，保證最終一致性。該機制使用了Data- serializing Operator（SUnion）來確保全部的副本處理一樣順序的數據。當失效恢復後，經過Checkpoint/Redo、Undo/Redo來實現恢 復重放。

對比

表1就上述3個流式系統作個分類比較，比較項基於DEBS2011會議上IFPSurvey中涉及的各類Models。Processing Model描述流元組進行計算時的選擇策略、消費策略及負載降級處理。Interaction Model描述輸入組件和計算系統、計算系統內部及計算系統和輸出組件的交互方式。Time Model描述事件流是否按照時間約束。Rules Model描述流式計算規則是顯示仍是隱式。Data Model描述流中的數據組成、格式等。Function Model描述流式計算系統的功能模型。Language Model描述語言層面的各類算子。

 

表1 3種流式系統的模型對比

小結
本文介紹了業界主流的3個流式計算系統，但願從這些系統的設計中領悟到不一樣場景下流式計算所要解決的關鍵問題。
Yahoo! S4的最新版本是Alpha version v0.3.0，動態負載均衡和在線服務遷移等重要功能都還沒有實現，不過其表明性的3個特色值得學習，Actor模式、非中心化的對稱結構及可插入式的架構。

StreamBase是有着功能強大的IDE而且支持控件式的方法來搭建應用程序，同時還提供了高級語言來搭建應用程序的方法。因爲是商業產品，其用戶接口的精彩設計值得借鑑，同時其可組合的HA方案也是亮點之一。 Borealis是學術界研究的重要產出，它對新一代的流式系統涉及的諸多方面，如係數據模型、負載管理、高可用性、可擴展性都做了全面和翔實的研究，一 方面系統變得強大、先進，另外一方面使得系統也變得臃腫、複雜。這套系統的許多策略都值得咱們學習，能夠應用於不一樣的流式計算場景。