作者 |?吳惠君，呂能，符茂松

責編 | 郭芮

【導語】?本文對比了Heron和常見的流處理項目，包括Storm、Flink、Spark Streaming和Kafka Streams，歸納了系統選型的要點。此外實踐了Heron的一個案例，以及討論了Heron在這一年開發的新特性。

在今年6月期的“基礎篇”中，我們通過學習Heron[1][2][3]的基本概念、整體架構和核心組件等內容，對Heron的設計、運行等方面有了基本的了解。在這一期的“應用篇”中，我們將Heron與其他流行的實時流處理系統（Apache Storm[4][5]、Apache Flink[6]、Apache Spark Streaming[7]和Apache Kafka Streams[8]）進行比較。在此基礎上，我們再介紹如何在實際應用中進行系統選型。然后我們將分享一個簡單的案例應用。最后我們會介紹在即將完結的2017年里Heron有哪些新的進展。

實時流處理系統比較與選型

當前流行的實時流處理系統主要包括Apache基金會旗下的Apache Storm、Apache Flink、Apache Spark Streaming和Apache Kafka Streams等項目。雖然它們和Heron同屬于實時流處理范疇，但是它們也有各自的特點。

Heron對比Storm（包括Trident）

在Twitter內部，Heron替換了Storm，是流處理的標準。

數據模型的區別

Heron兼容Storm的數據模型，或者說Heron兼容Storm的API，但是背后的實現完全不同。所以它們的應用場景是一樣的，能用Storm的地方也能用Heron。但是Heron比Storm提供更好的效率，更多的功能，更穩定，更易于維護。

Storm Trident是Storm基礎上的項目，提供高級別的API，如同Heron的函數式API。Trident以checkpoint加rollback的方式實現了exactly once；Heron以Chandy和Lamport發明的分布式快照算法實現了effectively once。

應用程序架構的區別

Storm的worker在每個JVM進程中運行多個線程，每個線程中執行多個任務。這些任務的log混在一起，很難調試不同任務的性能。Storm的nimbus無法對worker進行資源隔離，所以多個topology的資源之間互相影響。另外ZooKeeper被用來管理heartbeat，這使得ZooKeeper很容易變成瓶頸。

Heron的每個任務都是單獨的JVM進程，方便調試和資源隔離管理，同時節省了整個topology的資源。ZooKeeper在Heron中只存放很少量的數據，heartbeat由tmaster進程管理，對ZooKeeper沒有壓力。

Heron對比Flink

Flink框架包含批處理和流處理兩方面的功能。Flink的核心采用流處理的模式，它的批處理模式通過模擬塊數據的的流處理形式得到。

數據模型的區別

Flink在API方面采用declarative的API模式。Heron既提供declarative模式API或者叫做functional API也提供底層compositional模式的API，此外Heron還提供Python[9]和C++[10]的API。

應用程序架構的區別

在運行方面，Flink可以有多種配置，一般情況采用的是多任務多線程在同一個JVM中的混雜模式，不利于調試。Heron采用的是單任務單JVM的模式，利于調試與資源分配。

在資源池方面，Flink和Heron都可以與多種資源池合作，包括Mesos/Aurora、YARN、Kubernetes等。

Heron對比Spark Streaming

Spark Streaming處理tuple的粒度是micro-batch，通常使用半秒到幾秒的時間窗口，將這個窗口內的tuple作為一個micro-batch提交給Spark處理。而Heron使用的處理粒度是tuple。由于時間窗口的限制，Spark Streaming的平均響應周期可以認為是半個時間窗口的長度，而Heron就沒有這個限制。所以Heron是低延遲，而Spark Streaming是高延遲。

Spark Streaming近期公布了一項提案，計劃在下一個版本2.3中加入一個新的模式，新的模式不使用micro-batch來進行計算。

數據模型的區別

語義層面上，Spark Streaming和Heron都實現了exactly once/effectively once。狀態層面上，Spark Streaming和Heron都實現了stateful processing。API接口方面，Spark Streaming支持SQL，Heron暫不支持。Spark Streaming和Heron都支持Java、Python接口。需要指出的是，Heron的API是pluggable模式的，除了Java和Python以外，Heron可以支持許多編程語言，比如C++。

應用程序架構的區別

任務分配方面，Spark Streaming對每個任務使用單個線程。一個JVM進程中可能有多個任務的線程在同時運行。Heron對每個任務都是一個單獨的heron-instance進程，這樣的設計是為了方便調試，因為當一個task失敗的時候，只用把這個任務進程拿出來檢查就好了，避免了進程中各個任務線程相互影響。

資源池方面，Spark Streaming和Heron都可以運行在YARN和Mesos上。需要指出的是Heron的資源池設計是pluggable interface的模式，可以連接許多資源管理器，比如Aurora等。讀者可以查看[11]了解Heron支持的資源池。

Heron對比Kafka Streams

Kafka Streams是一個客戶端的程序庫。通過這個調用庫，應用程序可以讀取Kafka中的消息流進行處理。

數據模型的區別

Kafka Streams與Kafka綁定，需要訂閱topic來獲取消息流，這與Heron的DAG模型完全不同。對于DAG模式的流計算，DAG的結點都是由流計算框架控制，用戶計算邏輯需要按照DAG的模式提交給這些框架。Kafka Streams沒有這些預設，用戶的計算邏輯完全用戶控制，不必按照DAG的模式。此外，Kafka Streams也支持反壓（back pressure）和stateful processing。

Kafka Streams定義了2種抽象：KStream和KTable。在KStream中，每一對key-value是獨立的。在KTable中，key-value以序列的形式解析。

應用程序架構的區別

Kafka Streams是完全基于Kafka來建設的，與Heron等流處理系統差別很大。Kafka Streams的計算邏輯完全由用戶程序控制，也就是說流計算的邏輯并不在Kafka集群中運行。Kafka Streams可以理解為一個連接器，從Kafka集群中讀取和寫入鍵值序列，計算所需資源和任務生命周期等等都要用戶程序管理。而Heron可以理解為一個平臺，用戶提交topology以后，剩下的由Heron完成。

選型

歸納以上對各個系統的比較，我們可以得到如上的表基于以上表格的比較，我們可以得到如下的選型要點：

表1 各系統比較

• Storm適用于需要快速響應、中等流量的場景。Storm和Heron在API上兼容，在功能上基本可以互換；Twitter從Storm遷移到了Heron，說明如果Storm和Heron二選一的話，一般都是選Heron。

• Kafka Streams與Kafka綁定，如果現有系統是基于Kafka構建的，可以考慮使用Kafka Streams，減少各種開銷。

• 一般認為Spark Streaming的流量是這些項目中最高的，但是它的響應延遲也是最高的。對于響應速度要求不高、但是對流通量要求高的系統，可以采用Spark Streaming；如果把這種情況推廣到極致就可以直接使用Spark系統。

• Flink使用了流處理的內核，同時提供了流處理和批處理的接口。如果項目中需要同時兼顧流處理和批處理的情況，Flink比較適合。同時因為需要兼顧兩邊的取舍，在單個方面就不容易進行針對性的優化和處理。

總結上面，Spark Streaming、Kafka Streams、Flink都有特定的應用場景，其他一般流處理情況下可以使用Heron。

Heron案例學習

讓我們在Ubuntu單機上來實踐運行一個示例topology，這包括如下幾個步驟：

• 安裝Heron客戶端，?
  啟動一個Heron示例topology，?
  其他topology操作命令。

• 安裝Heron工具包，?
  運行Heron Tracker，?
  運行Heron UI。

運行topology

首先找到Heron的發布網頁：https://github.com/twitter/heron/releases，找到最新的版本0.16.5。可以看到Heron提供了多個版本的安裝文件，這些安裝文件又分為幾個類別：客戶端client、工具包tools和開發包API等。

安裝客戶端

下載客戶端安裝文件heron-client-install-0.16.5-ubuntu.sh：

wget https://github.com/twitter/heron/releases/download/0.16.5/heron-client-install-0.16.5-darwin.sh

然后執行這個文件：

chmod +x heron-*.sh

./heron-client-install-0.16.5--PLATFORM.sh --user

其中--user參數讓heron客戶端安裝到當前用戶目錄~/.hedon，同時在~/bin下創建一個鏈接指向~/.heorn/bin下的可執行文件。

Heron客戶端是一個名字叫heron的命令行程序。可以通過export PATH=~/bin:$PATH讓heron命令能被直接訪問。運行如下命令來檢測heron命令是否安裝成功：

heron version

運行示例topology

首先添加localhost到/etc/hosts，Heron在單機模式時會用/etc/hosts來解析本地域名。

Heron客戶端安裝時已經包含了一個示例topology的jar包，在~/.heron/example目錄下。我們可以運行其中一個示例topology作為例子：

heron submit local ~/.heron/examples/heron-examples.jar \com.twitter.heron.examples.ExclamationTopology ExclamationTopology \--deploy-deactivated

heron submit命令提交一個topology給heron運行。關于heron submit的命令的格式，可以用過heron help submit來查看。

當Heron運行在單機本地模式時，它會將運行狀態和日志等信息存放在~/.herondata目錄下。我們可以可以查看剛才運行的示例topology目錄，具體位置是：

ls -al ~/.herondata/topologies/local/${USER_NAME}/ExclamationTopology

Topology生命周期

一個topology的生命周期包括如下幾個階段：

• submit：提交topology給heron-scheduler。這時topology還沒有處理tuples，但是它已經準備好，等待被activate；

• activate/deactivate：讓topology開始/停止處理tuples；

• restart：重啟一個topology，讓資源管理器重新分配容器；

• kill：撤銷topology， 釋放資源。

這些階段都是通過heron命令行客戶端來管理的。具體的命令格式可以通過heron help查看。

Heron工具包

Heron項目提供了一些工具，可以方便查看數據中心中運行的topology狀態。在單機本地模式下，我們也可以來試試這些工具。這些工具主要包括：

• Tracker：一個服務器提供restful API，監視每個topology的運行時狀態；

• UI：一個網站，調用Tracker restful API展示成網頁。

一個數據中心內可以部署一套工具包來涵蓋整個數據中心的所有topology。

安裝工具包

用安裝Heron客戶端類似的方法，找到安裝文件，然后安裝它：

wget https://github.com/twitter/heron/releases/download/0.16.5/heron-tools-install-0.16.5-darwin.sh

chmod +x heron-*.sh

./heron-tools-install-0.16.5-PLATFORM.sh --user

Tracker工具

啟動Tracker服務器：heron-tracker

驗證服務器restful api：在瀏覽器中打開http://localhost:8888

圖1 啟動Tracker服務器

UI 工具

啟動UI網站：heron-ui

驗證UI網站：在瀏覽器中打開http://localhost:8889

圖2 啟動UI網站

Heron新特性

自從2016年夏Twitter開源Heron以來，Heron社區開發了許多新的功能，特別是2017年Heron增加了“在線動態擴容／縮容”、“effectively once傳輸語義”、“函數式API”、“多種編程語言支持”、“自我調節（self-regulating）”等。

在線動態擴容／縮容

根據Storm的數據模型，topology的并行度是topology的作者在編程topology的時候指定的。很多情況下，topology需要應付的數據流量在不停的變化。topology的編程者很難預估適合的資源配置，所以動態的調整topology的資源配置就是運行時的必要功能需求。

直觀地，改變topology中結點的并行度就能快速改變topology的資源使用量來應付數據流量的變換。Heron通過update命令來實現這種動態調整。Heron命令行工具使用packing算法按照用戶指定的新的并行度計算topology的新的packing plan，然后通過資源池調度器增加或者減少容器數量，并再將這個packing plan發送給tmaster合并成新的physical plan，使得整個topology所有容器狀態一致。Heron實現的并行度動態調整對運行時的topology影響小，調整快速。

Effectively once傳輸語義

Heron在原有tuple傳輸模式at most once和at least once以外，新加入了effectively once。原有的at most once和at least once都有些不足之處，比如at most once會漏掉某些tuple；而at least once會重復某些tuple。所以effectively once的目標是，當計算是確定性（deterministic）的時候，結果精確可信。

Effectively once的實現可以概括為兩點：

• 分布式狀態checkpoint；

• topology狀態回滾。

tmaster定期向spout發送marker tuple。當topology中的一個結點收集齊上游的marker tuple時，會將當時自己的狀態寫入一個state storage，這個過程就是checkpoint。當整個topology的所有結點都完成checkpoint的時候，state storage就存儲了一份整個topology快照。如果topology遇到異常，可以從state storage讀取快照進行恢復并重新開始處理數據。

函數式API （Functional API）

函數式編程是近年來的熱點，Heron適應時代潮流在原有API的基礎上添加了函數式API。Heron的函數式API讓topology編程者更專注于topology的應用邏輯，而不必關心topology/spout/bolt的具體細節。Heron的函數式API相比于原有的底層API是一種更高層級上的API，它背后的實現仍然是轉化為底層API來構建topology。

Heron函數式API建立在streamlet的概念上。一個streamlet是一個無限的、順序的tuple序列。Heron函數式API的數據模型中，數據處理就是指從一個streamlet轉變為另一個streamlet。轉變的操作包括：map、flatmap、join、filter和window等常見的函數式操作。

多種編程語言支持

以往topology編寫者通常使用兼容Storm的Java API來編寫topology，現在Heron提供Python和C++的API，讓熟悉Python和C++的程序員也可以編寫topology。Python和C++的API設計與Java API類似，它們包含底層API用來構造DAG，將來也會提供函數式API讓topology開發者更專注業務邏輯。

在實現上，Python和C++的API都有Python和C++的heron-instance實現。它們不與heron-instance的Java實現重疊，所以減少了語言間轉化的開銷，提高了效率。

自我調節（self-regulating）

Heron結合Dhalion框架開發了新的health manager模塊。Dhalion框架是一個讀取metric然后對topology進行相應調整或者修復的框架。Health manager由2個部分組成：detector/diagnoser和resolver。Detector/diagnoser讀取metric探測topology狀態并發現異常，resolver根據發現的異常執行相應的措施讓topology恢復正常。Health manager模塊的引入，讓Heron形成了完整的反饋閉環。

現在常用的兩個場景是：1. detector監測back pressure和stmgr中隊列的長度，發現是否有些容器是非常慢的；然后resolver告知heron-scheduler來重新調度這個結點到其他host上去；2. detector監測所有結點的狀態來計算topology在全局層面上是不是資源緊張，如果發現topology資源使用量很大，resolver計算需要添加的資源并告知scheduler來進行調度。

結束語

在本文中，我們對比了Heron和常見的流處理項目，包括Storm、Flink、Spark Streaming和Kafka Streams，歸納了系統選型的要點，此外我們實踐了Heron的一個案例，最后我們討論了Heron在這一年開發的新特性。

最后，作者希望這篇文章能為大家提供一些Heron應用的相關經驗，也歡迎大家向我們提供建議和幫助。如果大家對Heron的開發和改進感興趣，可以查看Heron官網（http://heronstreaming.io）和代碼（https://github.com/twitter/heron）。

參考文獻

[1] Maosong Fu, Ashvin Agrawal, Avrilia Floratou, Bill Graham, Andrew Jorgensen, Mark Li, Neng Lu, Karthik Ramasamy, Sriram Rao, and Cong Wang. “Twitter Heron: Towards Extensible Streaming Engines.” In 2017 IEEE 33rd International Conference on Data Engineering (ICDE), pp. 1165-1172. IEEE, 2017.?
[2] Sanjeev Kulkarni, Nikunj Bhagat, Maosong Fu, Vikas Kedigehalli, Christopher Kellogg, Sailesh Mittal, Jignesh M. Patel, Karthik Ramasamy, and Siddarth Taneja. “Twitter heron: Stream processing at scale.” In Proceedings of the 2015 ACM SIGMOD International Conference on Management of Data, pp. 239-250. ACM, 2015.?
[3] Maosong Fu, Sailesh Mittal, Vikas Kedigehalli, Karthik Ramasamy, Michael Barry, Andrew Jorgensen, Christopher Kellogg, Neng Lu, Bill Graham, and Jingwei Wu. “Streaming@ Twitter.” IEEE Data Eng. Bull. 38, no. 4 (2015): 15-27.?
[4]?http://storm.apache.org/?
[5]?http://storm.apache.org/releases/current/Trident-tutorial.html?
[6]?https://flink.apache.org/?
[7]?https://spark.apache.org/streaming/?
[8]?https://kafka.apache.org/documentation/streams/?
[9]?https://twitter.github.io/heron/api/python/?
[10]?https://github.com/twitter/heron/tree/master/heron/instance/src/cpp?
[11]?https://github.com/twitter/heron/tree/master/heron/schedulers/src/java/com/twitter/heron/scheduler

作者簡介：吳惠君，Twitter軟件工程師，致力于實時流處理引擎Heron的研究和開發。他畢業于Arizona State University，專攻大數據處理和移動云計算，曾在國際頂級期刊和會議發表多篇學術論文，并有多項專利。?
呂能，Twitter實時計算平臺團隊成員。專注于分布式系統，曾參與過Twitter的Manhattan鍵值存儲系統，Obs監控警報系統的開發，目前負責Heron的開發研究。曾在國際頂級期刊和會議發表多篇學術論文。?
符茂松，Twitter實時計算平臺團隊主管，負責Heron, Presto等服務。Heron的原作者之一。專注于分布式系統，在SIGMOD, ICDE等會議期刊發表多篇論文。本科畢業于華中科技大學；研究生畢業于Carnegie Mellon University。?
本文為《程序員》原創文章，未經允許不得轉載。

1月13日，SDCC 2017之數據庫線上峰會即將強勢來襲，秉承干貨實料（案例）的內容原則，邀請了來自阿里巴巴、騰訊、微博、網易等多家企業的數據庫專家及高校研究學者，圍繞Oracle、MySQL、PostgreSQL、Redis等熱點數據庫技術展開，從核心技術的深挖到高可用實踐的剖析，打造精華壓縮式分享，舉一反三，思辨互搏，報名及更多詳情可點擊「閱讀原文」查看。

總結

以上是生活随笔為你收集整理的Heron：来自Twitter的新一代流处理引擎应用篇的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。