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

責(zé)編 | 郭芮

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

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

實時流處理系統(tǒng)比較與選型

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

Heron對比Storm（包括Trident）

在Twitter內(nèi)部，Heron替換了Storm，是流處理的標(biāo)準(zhǔn)。

數(shù)據(jù)模型的區(qū)別

Heron兼容Storm的數(shù)據(jù)模型，或者說Heron兼容Storm的API，但是背后的實現(xiàn)完全不同。所以它們的應(yīng)用場景是一樣的，能用Storm的地方也能用Heron。但是Heron比Storm提供更好的效率，更多的功能，更穩(wěn)定，更易于維護(hù)。

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

應(yīng)用程序架構(gòu)的區(qū)別

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

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

Heron對比Flink

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

數(shù)據(jù)模型的區(qū)別

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

應(yīng)用程序架構(gòu)的區(qū)別

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

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

Heron對比Spark Streaming

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

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

數(shù)據(jù)模型的區(qū)別

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

應(yīng)用程序架構(gòu)的區(qū)別

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

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

Heron對比Kafka Streams

Kafka Streams是一個客戶端的程序庫。通過這個調(diào)用庫，應(yīng)用程序可以讀取Kafka中的消息流進(jìn)行處理。

數(shù)據(jù)模型的區(qū)別

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

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

應(yīng)用程序架構(gòu)的區(qū)別

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

選型

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

表1 各系統(tǒng)比較

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

• Kafka Streams與Kafka綁定，如果現(xiàn)有系統(tǒng)是基于Kafka構(gòu)建的，可以考慮使用Kafka Streams，減少各種開銷。

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

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

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

Heron案例學(xué)習(xí)

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

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

• 安裝Heron工具包，?
  運(yùn)行Heron Tracker，?
  運(yùn)行Heron UI。

運(yùn)行topology

首先找到Heron的發(fā)布網(wǎng)頁：https://github.com/twitter/heron/releases，找到最新的版本0.16.5。可以看到Heron提供了多個版本的安裝文件，這些安裝文件又分為幾個類別：客戶端client、工具包tools和開發(fā)包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

然后執(zhí)行這個文件：

chmod +x heron-*.sh

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

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

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

heron version

運(yùn)行示例topology

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

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

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

heron submit命令提交一個topology給heron運(yùn)行。關(guān)于heron submit的命令的格式，可以用過heron help submit來查看。

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

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

Topology生命周期

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

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

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

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

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

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

Heron工具包

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

• Tracker：一個服務(wù)器提供restful API，監(jiān)視每個topology的運(yùn)行時狀態(tài)；

• UI：一個網(wǎng)站，調(diào)用Tracker restful API展示成網(wǎng)頁。

一個數(shù)據(jù)中心內(nèi)可以部署一套工具包來涵蓋整個數(shù)據(jù)中心的所有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服務(wù)器：heron-tracker

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

圖1 啟動Tracker服務(wù)器

UI 工具

啟動UI網(wǎng)站：heron-ui

驗證UI網(wǎng)站：在瀏覽器中打開http://localhost:8889

圖2 啟動UI網(wǎng)站

Heron新特性

自從2016年夏Twitter開源Heron以來，Heron社區(qū)開發(fā)了許多新的功能，特別是2017年Heron增加了“在線動態(tài)擴(kuò)容／縮容”、“effectively once傳輸語義”、“函數(shù)式API”、“多種編程語言支持”、“自我調(diào)節(jié)（self-regulating）”等。

在線動態(tài)擴(kuò)容／縮容

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

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

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會重復(fù)某些tuple。所以effectively once的目標(biāo)是，當(dāng)計算是確定性（deterministic）的時候，結(jié)果精確可信。

Effectively once的實現(xiàn)可以概括為兩點：

• 分布式狀態(tài)checkpoint；

• topology狀態(tài)回滾。

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

函數(shù)式API （Functional API）

函數(shù)式編程是近年來的熱點，Heron適應(yīng)時代潮流在原有API的基礎(chǔ)上添加了函數(shù)式API。Heron的函數(shù)式API讓topology編程者更專注于topology的應(yīng)用邏輯，而不必關(guān)心topology/spout/bolt的具體細(xì)節(jié)。Heron的函數(shù)式API相比于原有的底層API是一種更高層級上的API，它背后的實現(xiàn)仍然是轉(zhuǎn)化為底層API來構(gòu)建topology。

Heron函數(shù)式API建立在streamlet的概念上。一個streamlet是一個無限的、順序的tuple序列。Heron函數(shù)式API的數(shù)據(jù)模型中，數(shù)據(jù)處理就是指從一個streamlet轉(zhuǎn)變?yōu)榱硪粋€streamlet。轉(zhuǎn)變的操作包括：map、flatmap、join、filter和window等常見的函數(shù)式操作。

多種編程語言支持

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

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

自我調(diào)節(jié)（self-regulating）

Heron結(jié)合Dhalion框架開發(fā)了新的health manager模塊。Dhalion框架是一個讀取metric然后對topology進(jìn)行相應(yīng)調(diào)整或者修復(fù)的框架。Health manager由2個部分組成：detector/diagnoser和resolver。Detector/diagnoser讀取metric探測topology狀態(tài)并發(fā)現(xiàn)異常，resolver根據(jù)發(fā)現(xiàn)的異常執(zhí)行相應(yīng)的措施讓topology恢復(fù)正常。Health manager模塊的引入，讓Heron形成了完整的反饋閉環(huán)。

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

結(jié)束語

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

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

參考文獻(xiàn)

[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的研究和開發(fā)。他畢業(yè)于Arizona State University，專攻大數(shù)據(jù)處理和移動云計算，曾在國際頂級期刊和會議發(fā)表多篇學(xué)術(shù)論文，并有多項專利。?
呂能，Twitter實時計算平臺團(tuán)隊成員。專注于分布式系統(tǒng)，曾參與過Twitter的Manhattan鍵值存儲系統(tǒng)，Obs監(jiān)控警報系統(tǒng)的開發(fā)，目前負(fù)責(zé)Heron的開發(fā)研究。曾在國際頂級期刊和會議發(fā)表多篇學(xué)術(shù)論文。?
符茂松，Twitter實時計算平臺團(tuán)隊主管，負(fù)責(zé)Heron, Presto等服務(wù)。Heron的原作者之一。專注于分布式系統(tǒng)，在SIGMOD, ICDE等會議期刊發(fā)表多篇論文。本科畢業(yè)于華中科技大學(xué)；研究生畢業(yè)于Carnegie Mellon University。?
本文為《程序員》原創(chuàng)文章，未經(jīng)允許不得轉(zhuǎn)載。

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

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。