流計算又稱實時計算，是繼以Map-Reduce為代表的批處理之后的又一重要計算模型。隨著互聯網業務的發展以及數據規模的持續擴大，傳統的批處理計算難以有效地對數據進行快速低延遲處理并返回結果。由于數據幾乎處于不斷增長的狀態中，及時處理計算大批量數據成為了批處理計算的一大難題。在此背景之下，流計算應運而生。相比于傳統的批處理計算，流計算具有低延遲、高響應、持續處理的特點。在數據產生的同時，就可以進行計算并獲得結果。更可以通過Lambda架構將即時的流計算處理結果與延后的批處理計算結果結合，從而較好地滿足低延遲、高正確性的業務需求。

Twitter由于本身的業務特性，對實時性有著強烈的需求。因此在流計算上投入了大量的資源進行開發。第一代流處理系統Storm發布以后得到了廣泛的關注和應用。根據Storm在實踐中遇到的性能、規模、可用性等方面的問題，Twitter又開發了第二代流處理系統——Heron[1]，并在2016年將它開源。

重要概念定義

在開始了解Heron的具體架構和設計之前，我們首先定義一些流計算以及在Heron設計中用到的基本概念：

• Tuple：流計算任務中處理的最小單元數據的抽象。

• Stream：由無限個Tuple組成的連續序列。

• Spout：從外界數據源獲得數據并生成Tuple的計算任務。

• Bolt：處理上游Spout或者Bolt生成的Tuple的計算任務。

• Topology：一個通過Stream將Spout和Bolt相連的處理Tuple的邏輯計算任務。

• Grouping：流計算中的Tuple分發策略。在Tuple通過Stream傳遞到下游Bolt的過程中，Grouping策略決定了如何將一個Tuple路由給一個具體的Bolt實例。典型的Grouping策略有：隨機分配、基于Tuple內容的分配等。

• Physical Plan：基于Topology定義的邏輯計算任務以及所擁有的計算資源，生成的實際運行時信息的集合。

在以上流處理基本概念的基礎上，我們可以構建出流處理的三種不同處理語義：

• 至多一次（At-Most-Once）： 盡可能處理數據，但不保證數據一定會被處理。吞吐量大，計算快但是計算結果存在一定的誤差。

• 至少一次（At-Least-Once）：在外部數據源允許Replay（重演）的情況下，保證數據至少被處理一次。在出現錯誤的情況下會重新處理該數據，可能會出現重復處理多次同一數據的情況。保證數據的處理但是延遲升高。

• 僅有一次（Exactly-Once）：每一個數據確保被處理且僅被處理一次。結果精確但是所需要的計算資源增多并且還會導致計算效率降低。

從上可知，三種不同的處理模式有各自的優缺點，因此在選擇處理模式的時候需要綜合考量一個Topology對于吞吐量、延遲、結果誤差、計算資源的要求，從而做出最優的選擇。目前的Heron已經實現支持至多一次和至少一次語義，并且正在開發對于僅有一次語義的支持。

Heron系統概覽

保持與Storm接口（API）兼容是Heron的設計目標之一。因此，Heron的數據模型與Storm的數據模型基本保持一致。每個提交給Heron的Topology都是一個由Spout和Bolt這兩類結點（Vertex）組成的，以Stream為邊（Edge）的有向無環圖（Directed acyclic graph）。其中Spout結點是Topology的數據源，它從外部讀取Topology所需要處理的數據，常見的如kafka-spout，然后發送給后續的Bolt結點進行處理。Bolt節點進行實際的數據計算，常見的運算如Filter、Map以及FlatMap等。

我們可以把Heron的Topology類比為數據庫的邏輯查詢計劃。這種邏輯上的計劃最后都要變成實質上的處理計劃才能執行。用戶在編寫Topology時指定每個Spout和Bolt任務的并行度和Tuple在Topology中結點間的分發策略（Grouping）。所有用戶提供的信息經過打包算法（Pakcing）的計算，這些Spout和Bolt任務（task）被分配到一批抽象容器中。最后再把這些抽象容器映射到真實的容器中，就可以生成一個物理上可執行的計劃（Physical plan），它是所有邏輯信息（拓撲圖、并行度、計算任務）和運行時信息（計算任務和容器的對應關系、實際運行地址）的集合。

整體結構

總體上，Heron的整體架構如圖1所示。用戶通過命令行工具（Heron-CLI）將Topology提交給Heron Scheduler。再由Scheduler對提交的Topology進行資源分配以及運行調度。在同一時間，同一個資源平臺上可以運行多個相互獨立Topology。

圖1 Heron架構

與Storm的Service架構不同，Heron是Library架構。Storm在架構設計上是基于服務的，因此需要設立專有的Storm集群來運行用戶提交的Topology。在開發、運維以及成本上，都有諸多的不足。而Heron則是基于庫的，可以運行在任意的共享資源調度平臺上。最大化地降低了運維負擔以及成本開銷。

目前的Heron支持Aurora、YARN、Mesos以及EC2，而Kubernetes和Docker等目前正在開發中。通過可擴展插件Heron Scheduler，用戶可以根據不同的需求及實際情況選擇相應的運行平臺，從而達到多平臺資源管理器的支持[2]。

而被提交運行Topology的內部結構如圖2所示，不同的計算任務被封裝在多個容器中運行。這些由調度器調度的容器可以在同一個物理主機上，也可分布在多個主機上。其中每一個Topology的第一個容器（容器0）負責整個Topology的管理工作，主要運行一個Topology Master進程；其余各個容器負責用戶提交的計算邏輯的實現，每個容器中主要運行一個Stream Manager進程，一個Metrics Manager進程，以及多個Instance進程。每個Instance都負責運行一個Spout或者Bolt任務（task）。對于Topology Master、Stream Manager以及Instance進程的結構及重要功能，我們會在本文的后面章節進行詳細的分析。

圖2 Topology結構

狀態（State）存儲和監控

Heron的State Manager是一個抽象的模塊，它在具體實現中可以是ZooKeeper或者是文件系統。它的主要作用是保存各個Topology的各種元信息：Topology的提交者、提交時間、運行時生成的Physical Plan以及Topology Master的地址等，從而為Topology的自我恢復提供幫助。

每個容器中的Metrics Manager負責收集所在容器的運行時狀態指標（Metrics），并上傳給監控系統。當前Heron版本中，簡化的監控系統集成在Topology Master中。將來這一監控模塊將會成為容器0中的一個獨立進程。Heron還提供Heron-Tracker和Heron-UI 這兩個工具來查看和監測一個數據中心中運行的所有Topology。

啟動過程

在一個Topology中，Topology Master是整個Topology的元信息管理者，它維護著完整的Topology元信息。而Stream Manager是每個容器的網關，它負責各個Instance之間的數據通信，以及和Topology Master之間的控制信令。

當用戶提交Topology之后，Scheduler便會開始分配資源并運行容器。每個容器中啟動一個Heron Executor的進程，它區分容器0和其他容器，分別啟動Topology Master或者Stream Manager等進程。在一個普通容器中，Instance進程啟動后會主動向本地容器的Stream Manager進行注冊。當Stream Manager收到所有Instance的注冊請求后，會向Topology Master發送包含了自己的所負責的Instance的注冊信息。當Topology Master收到所有Stream Manager的注冊信息以后，會生成一個各個Instance，Stream Manager的實際運行地址的Physical Plan并進行廣播分發。收到了Physical Plan的各個Stream Manager之間就可以根據這一Physical Plan互相建立連接形成一個完全圖，然后開始處理數據。

Instance進行具體的Tuple數據計算處理。Stream Manager則不執行具體的計算處理任務，只負責中繼轉發Tuple。從數據流網絡的角度，可以把Stream Manager理解為每個容器的路由器。所有Instance之間的Tuple傳遞都是通過Stream Manager中繼。因此容器內的Instance之間通信是一跳（hop）的星形網絡。所有的Stream Manager都互相連接，形成Mesh網絡。容器之間的通信也是通過Stream Manager中繼的，是通過兩跳的中繼完成的。

核心組件分析

TMaster

TMaster是Topology Master的簡寫。與很多Master-Slave模式分布式系統中的Master單點處理控制邏輯的作用相同，TMaster作為Master角色提供了一個全局的接口來了解Topology的運行狀態。同時，通過將重要的狀態信息（Physical Plan）等記錄到ZooKeeper中，保證了TMaster在崩潰恢復之后能繼續運行。

實際產品中的TMaster在啟動的時候，會在ZooKeeper的某一約定目錄中創建一個Ephemeral Node來存儲自己的IP地址以及端口，讓Stream Manager能發現自己。Heron使用Ephemeral Node的原因包括：

• 避免了一個Topology出現多個TMaster的情況。這樣就使得這個Topology的所有進程都能認定同一個TMaster；

• 同一Topology內部的進程能夠通過ZooKeeper來發現TMaster所在的位置，從而與其建立連接。

TMaster主要有以下三個功能：

• 構建、分發并維護Topology的Physical Plan；

• 收集各個Stream Manager的心跳，確認Stream Manager的存活；

• 收集和分發Topology部分重要的運行時狀態指標（Metrics）。

由于Topology的Physical Plan只有在運行時才能確定，因此TMaster就成為了構建、分發以及維護Physical Plan的最佳選擇。在TMaster完成啟動和向ZooKeeper注冊之后，會等待所有的Stream Manager與自己建立連接。在Stream Manager與TMaster建立連接之后，Stream Manager會報告自己的實際IP地址、端口以及自己所負責的Instance地址與端口。TMaster在收到所有Stream Manager報告的地址信息之后就能構建出Physical Plan并進行廣播分發。所有的Stream Manager都會收到由TMaster構建的Physical Plan，并且根據其中的信息與其余的Stream Manager建立兩兩連接。只有當所有的連接都建立完成之后，Topology才會真正開始進行數據的運算和處理。當某一個Stream Manager丟失并重連之后，TMaster會檢測其運行地址及端口是否發生了改變；若改變，則會及時地更新Physical Plan并廣播分發，使Stream Manager能夠建立正確的連接，從而保證整個Topology的正確運行。

TMaster會接受Stream Manager定時發送的心跳信息并且維護各個Stream Manager的最近一次心跳時間戳。心跳首先能夠幫助TMaster確認Stream Manager的存活，其次可以幫助其決定是否更新一個Stream Manager的連接并且更新Physical Plan。

TMaster還會接受由Metrics Manager發送的一部分重要Metrics并且向Heron-Tracker提供這些Metrics。Heron-Tracker可以通過這些Metrics來確定Topology的運行情況并使得Heron-UI能夠基于這些重要的Metrics來進行監控檢測。典型的Metrics有：分發Tuple的次數，計算Tuple的次數以及處于backpressure狀態的時間等。

非常值得注意的一點是，TMaster本身并不參與任何實際的數據處理。因此它也不會接受和分發任何的Tuple。這一設計使得TMaster本身邏輯清晰，也非常輕量，同時也為以后功能的拓展留下了巨大的空間。

Stream Manager 和反壓（Back pressure）機制

Stmgr是Stream Manager的簡寫。Stmgr管理著Tuple的路由，并負責中繼Tuple。當Stmgr拿到Physical Plan以后就能根據其中的信息知道與其余的Stmgr建立連接形成Mesh網絡，從而進行數據中繼以及Backpressure控制。Tuple傳遞路徑可以通過圖3來說明，圖3中容器1的Instance D（1D）要發送一個Tuple給容器4中的Instance C（4C），這個Tuple經過的路徑為：容器1的1D，容器1的Stmgr，容器4的Stmgr，容器4的4C。又比如從3A到3B的Tuple經過的路徑為：3A，容器3的Stmgr，3B。與Internet的路由機制對比，Heron的路由非常簡單，這得益于Stmgr之間兩兩相連，使得所有的Instance之間的距離不超過2跳。

圖3 Tuple發送路徑示例

Acking

Stmgr除了路由中繼Tuple的功能以外，它還負責確認（Acking）Tuple已經被處理。Acking的概念在Heron的前身Storm中已經存在。Acking機制的目的是為了實現At-Least-Once的語義。原理上，當一個Bolt實例處理完一個Tuple以后，這個Bolt實例發送一個特殊的Acking Tuple給這個bolt的上游Bolt實例或者Spout實例，向上游結點確認Tuple已經處理完成。這個過程層層向上游結點推進，直到Spout結點。實現上，當Acking Tuple經過Stmgr時候由異或（xor）操作標記Tuple，由異或操作的特性得知是否處理完成。當一個Spout實例在一定時間內還沒有收集到Acking Tuple，那么它將重發對應的數據Tuple。Heron的Acking機制的實現與它的前任Storm一致。

Back Pressure

Heron引入了反壓（Back Pressure）機制，來動態調整Tuple的處理速度以避免系統過載。一般來說，解決系統過載問題有三種策略：1. 放任不管；2. 丟棄過載數據；3. 請求減少負載。Heron采用了第三種策略，通過Backpressure機制來進行過載恢復，保證系統不會在過載的情況下崩潰。

Backpressure機制觸發過程如下：當某一個Bolt Instance處理速度跟不上Tuple的輸入速度時，會造成負責向該Instance轉發Tuple的Stmgr緩存不斷堆積。當緩存大小超過一個上限值（Hight Water Mark）時，該Stmgr會停止從本地的Spout中讀取Tuple并向Topology中的其他所有Stmgr發送一個“開始Backpressure”的信息。而其余的Stmgr在接收到這一消息時也會停止從他們所負責的Spout Instance處讀取并轉發Tuple。至此，整個Topology就不再從外界讀入Tuple而只處理堆積在內部的未處理Tuple。而處理的速度則由最慢的Instance來決定。在經過一定時間的處理以后，當緩存的大小減低到一個下限值（Low Water Mark）時，最開始發送“開始Backpressure”的Stmgr會再次發送“停止Backpressure”的信息，從而使得所有的Stmgr重新開始從Spout Instance讀取分發數據。而由于Spout通常是從具有允許重演（Replay）的消息隊列中讀取數據，因此即使凍結了也不會導致數據的丟失。

注意在Backpressure的過程中兩個重要的數值：上限值（High Water Mark）和下限值（Low Water Mark）。只有當緩存區的大小超過上限值時才會觸發Backpressure，然后一直持續到緩存區的大小減低到下限值時。這一設計有效地避免了一個Topology不停地在Backpressure狀態和正常狀態之間震蕩變化的情況發展，一定程度上保證了Topology的穩定。

Instance

Instance是整個Heron處理引擎的核心部分之一。Topology中不論是Spout類型結點還是Bolt類型結點，都是由Instance來實現的。不同于Storm的Worker設計，在當前的Heron中每一個Instance都是一個獨立的JVM進程，通過Stmgr進行數據的分發接受，完成用戶定義的計算任務。獨立進程的設計帶來了一系列的優點：便于調試、調優、資源隔離以及容錯恢復等。同時，由于數據的分發傳送任務已經交由Stmgr來處理，Instance可以用任何編程語言來進行實現，從而支持各種語言平臺。

Instance采用雙線程的設計，如圖4所示。一個Instance的進程包含Gateway以及Task Execution這兩個線程。Gateway線程主要控制著Instance與本地Stmgr和Metrics Manager之間的數據交換。通過TCP連接，Gateway線程：1. 接受由Stmgr分發的待處理Tuple；2. 發送經Task Execution處理的Tuple給Stmgr；3. 轉發由Task Execution線程產生的Metrics給Metrics Manager。不論是Spout還是Bolt，Gateway線程完成的任務都相同。

Task Execution線程的職責是執行用戶定義的計算任務。對于Spout和Bolt，Task Execution線程會相應地去執行open()和prepare()方法來初始化其狀態。如果運行的Instance是一個Bolt實例，那么Task Execution線程會執行execute()方法來處理接收到的Tuple；如果是Spout，則會重復執行nextTuple()方法來從外部數據源不停地獲取數據，生成Tuple，并發送給下游的Instance進行處理。經過處理的Tuple會被發送至Gateway線程進行下一步的分發。同時在執行的過程中，Task Execution線程會生成各種Metrics（tuple處理數量，tuple處理延遲等）并發送給Metrics Manager進行狀態監控。

圖4 Instance結構

Gateway線程和Task Execution線程之間通過三個單向的隊列來進行通信，分別是數據進入隊列、數據發送隊列以及Metrics發送隊列。Gateway線程通過數據進入隊列向Task Execution線程傳入Tuple；Task Execution通過數據發送隊列將處理完的Tuple發送給Gateway線程；Task Execution線程通過Metrics發送隊列將收集的Metric發送給Gateway線程。

總結

在本文中，我們介紹了流計算的背景和重要概念，并且詳細分析了Twitter目前的流計算引擎—— Heron的結構及重要組件。希望能借此為大家提供一些在設計和構建流計算系統時的經驗，也歡迎大家向我們提供建議和幫助。如果大家對Heron的開發和改進感興趣，可以在Github上進行查看。

【1】Kulkarni, Sanjeev, 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.

【2】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 International Conference on Data Engineering (ICDE). IEEE, 2017.

呂能，Twitter實時計算平臺團隊成員。專注于分布式系統，曾參與過Twitter的Manhattan鍵值存儲系統，Obs監控警報系統的開發，目前負責Heron的開發研究。曾在國際頂級期刊和會議發表多篇學術論文。?

吳惠君，Twitter軟件工程師，致力于實時流處理引擎Heron的研究和開發。他畢業于Arizona State University，專攻大數據處理和移動云計算，曾在國際頂級期刊和會議發表多篇學術論文，并有多項專利。?

符茂松，Twitter實時計算平臺團隊主管，負責Heron, Presto等服務。Heron的原作者之一。專注于分布式系統，在SIGMOD、ICDE等會議期刊發表多篇論文。本科畢業于華中科技大學，研究生畢業于Carnegie Mellon University。?

本文為《程序員》原創文章，未經允許禁止轉載

總結

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

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