360深度實踐：Flink與Storm協議級對比

作者 張馨予，360 大數據計算平台負責人。北京郵電大學碩士，2015年加入360系統部，一直致力於公司大數據計算平台的易用性、穩定性和性能優化的研發工作。目前主要負責Flink的研發，完成公司計算引擎的大一統。

本文從數據傳輸和數據可靠性的角度出發，對比測試了Storm與Flink在流處理上的性能，並對測試結果進行分析，給出在使用Flink時提高性能的建議。

Apache Storm、Apache Spark和Apache Flink都是開源社區中非常活躍的分散式計算平台，在很多公司可能同時使用著其中兩種甚至三種。對於實時計算來說，Storm與Flink的底層計算引擎是基於流的，本質上是一條一條的數據進行處理，且處理的模式是流水線模式，即所有的處理進程同時存在，數據在這些進程之間流動處理。而Spark是基於批量數據的處理，即一小批一小批的數據進行處理，且處理的邏輯在一批數據準備好之後才會進行計算。在本文中，我們把同樣基於流處理的Storm和Flink拿來做對比測試分析。

在我們做測試之前，調研了一些已有的大數據平台性能測試報告，比如，雅虎的Streaming-benchmarks，或者Intel的HiBench等等。除此之外，還有很多的論文也從不同的角度對分散式計算平台進行了測試。雖然這些測試case各有不同的側重點，但他們都用到了同樣的兩個指標，即吞吐和延遲。吞吐表示單位時間內所能處理的數據量，是可以通過增大並發來提高的。延遲代表處理一條數據所需要的時間，與吞吐量成反比關係。

在我們設計計算邏輯時，首先考慮一下流處理的計算模型。上圖是一個簡單的流計算模型，在Source中將數據取出，發往下游Task，並在Task中進行處理，最後輸出。對於這樣的一個計算模型，延遲時間由三部分組成：數據傳輸時間、Task計算時間和數據排隊時間。我們假設資源足夠，數據不用排隊。則延遲時間就只由數據傳輸時間和Task計算時間組成。而在Task中處理所需要的時間與用戶的邏輯息息相關，所以對於一個計算平台來說，數據傳輸的時間才更能反映這個計算平台的能力。因此，我們在設計測試Case時，為了更好的體現出數據傳輸的能力，Task中沒有設計任何計算邏輯。

在確定數據源時，我們主要考慮是在進程中直接生成數據，這種方法在很多之前的測試標準中也同樣有使用。這樣做是因為數據的產生不會受到外界數據源系統的性能限制。但由於在我們公司內部大部分的實時計算數據都來源於kafka，所以我們增加了從kafka中讀取數據的測試。

對於數據傳輸方式，可以分為兩種：進程間的數據傳輸和進程內的數據傳輸。

進程間的數據傳輸是指這條數據會經過序列化、網路傳輸和反序列化三個步驟。在Flink中，2個處理邏輯分布在不同的TaskManager上，這兩個處理邏輯之間的數據傳輸就可以叫做進程間的數據傳輸。Flink網路傳輸是採用的Netty技術。在Storm中，進程間的數據傳輸是worker之間的數據傳輸。早版本的storm網路傳輸使用的ZeroMQ，現在也改成了Netty。

進程內的數據傳輸是指兩個處理邏輯在同一個進程中。在Flink中，這兩個處理邏輯被Chain在了一起，在一個線程中通過方法調用傳參的形式進程數據傳輸。在Storm中，兩個處理邏輯變成了兩個線程，通過一個共享的隊列進行數據傳輸。

Storm和Flink都有各自的可靠性機制。在Storm中，使用ACK機制來保證數據的可靠性。而在Flink中是通過checkpoint機制來保證的，這是來源於chandy-lamport演算法。

事實上exactly-once可靠性的保證跟處理的邏輯和結果輸出的設計有關。比如結果要輸出到kafka中，而輸出到kafka的數據無法回滾，這就無法保證exactly-once。我們在測試的時候選用的at-least-once語義的可靠性和不保證可靠性兩種策略進行測試。

上圖是我們測試的環境和各個平台的版本。

上圖展示的是Flink在自產數據的情況下，不同的傳輸方式和可靠性的吞吐量：在進程內+不可靠、進程內+可靠、進程間+不可靠、進程間+可靠。可以看到進程內的數據傳輸是進程間的數據傳輸的3.8倍。是否開啟checkpoint機制對Flink的吞吐影響並不大。因此我們在使用Flink時，進來使用進程內的傳輸，也就是儘可能的讓運算元可以Chain起來。

那麼我們來看一下為什麼Chain起來的性能好這麼多，要如何在寫Flink代碼的過程中讓Flink的運算元Chain起來使用進程間的數據傳輸。

大家知道我們在Flink代碼時一定會創建一個env，調用env的disableOperatorChainning方法會使得所有的運算元都無法chain起來。我們一般是在debug的時候回調用這個方法，方便調試問題。

如果允許Chain的情況下，上圖中Source和mapFunction就會Chain起來，放在一個Task中計算。反之，如果不允許Chain，則會放到兩個Task中。

對於沒有Chain起來的兩個運算元，他們被放到了不同的兩個Task中，那麼他們之間的數據傳輸是這樣的：SourceFunction取到數據序列化後放入內存，然後通過網路傳輸給MapFunction所在的進程，該進程將數據方序列化後使用。

對於Chain起來的兩個運算元，他們被放到同一個Task中，那麼這兩個運算元之間的數據傳輸則是：SourceFunction取到數據後，進行一次深拷貝，然後MapFunction把深拷貝出來的這個對象作為輸入數據。

雖然Flink在序列化上做了很多優化，跟不用序列化和不用網路傳輸的進程內數據傳輸對比，性能還是差很多。所以我們儘可能的把運算元Chain起來。

不是任何兩個運算元都可以Chain起來的，要把運算元Chain起來有很多條件：第一，下游運算元只能接受一種上游數據流，比如Map接受的流不能是一條union後的流；其次上下游的並發數一定要一樣；第三，運算元要使用同一個資源Group，默認是一致的，都是default；第四，就是之前說的env中不能調用disableOperatorChainning方法，最後，上游發送數據的方法是Forward的，比如，開發時沒有調用rebalance方法，沒有keyby，沒有boardcast等。

對比一下自產數據時，使用進程內通信，且不保證數據可靠性的情況下，Flink與Storm的吞吐。在這種情況下，Flink的性能是Storm的15倍。Flink吞吐能達到2060萬條/s。不僅如此，如果在開發時調用了env.getConfig.enableObjectReuse方法，Flink的但並發吞吐能達到4090萬條/s。

當調用了enableObjectReuse方法後，Flink會把中間深拷貝的步驟都省略掉，SourceFunction產生的數據直接作為MapFunction的輸入。但需要特別注意的是，這個方法不能隨便調用，必須要確保下游Function只有一種，或者下游的Function均不會改變對象內部的值。否則可能會有線程安全的問題。

當對比在不同可靠性策略的情況下，Flink與Storm的表現時，我們發現，保證可靠性對Flink的影響非常小，但對Storm的影響非常大。總的來說，在保證可靠的情況下，Flink單並發的吞吐是Storm的15倍，而不保證可靠的情況下，Flink的性能是Storm的66倍。會產生這樣的結果，主要是因為Flink與Storm保證數據可靠性的機制不同。

而Storm的ACK機製為了保證數據的可靠性，開銷更大。

左邊的圖展示的是Storm的Ack機制。Spout每發送一條數據到Bolt，就會產生一條ack的信息給acker，當Bolt處理完這條數據後也會發送ack信息給acker。當acker收到這條數據的所有ack信息時，會回復Spout一條ack信息。也就是說，對於一個只有兩級（spout+bolt）的拓撲來說，每發送一條數據，就會傳輸3條ack信息。這3條ack信息則是為了保證可靠性所需要的開銷。

右邊的圖展示的是Flink的Checkpoint機制。Flink中Checkpoint信息的發起者是JobManager。它不像Storm中那樣，每條信息都會有ack信息的開銷，而且按時間來計算花銷。用戶可以設置做checkpoint的頻率，比如10秒鐘做一次checkpoint。每做一次checkpoint，花銷只有從Source發往map的1條checkpoint信息（JobManager發出來的checkpoint信息走的是控制流，與數據流無關）。與storm相比，Flink的可靠性機制開銷要低得多。這也就是為什麼保證可靠性對Flink的性能影響較小，而storm的影響確很大的原因。

最後一組自產數據的測試結果對比是Flink與Storm在進程間的數據傳輸的對比，可以看到進程間數據傳輸的情況下，Flink但並發吞吐是Storm的4.7倍。保證可靠性的情況下，是Storm的14倍。

上圖展示的是消費kafka中數據時，Storm與Flink的但並發吞吐情況。因為消費的是kafka中的數據，所以吞吐量肯定會收到kafka的影響。我們發現性能的瓶頸是在SourceFunction上，於是增加了topic的partition數和SourceFunction取數據線程的並發數，但是MapFunction的並發數仍然是1.在這種情況下，我們發現flink的瓶頸轉移到上游往下游發數據的地方。而Storm的瓶頸確是在下游收數據反序列化的地方。

之前的性能分析使我們基於數據傳輸和數據可靠性的角度出發，單純的對Flink與Storm計算平台本身進行了性能分析。但實際使用時，task是肯定有計算邏輯的，這就勢必更多的涉及到CPU，內存等資源問題。我們將來打算做一個智能分析平台，對用戶的作業進行性能分析。通過收集到的指標信息，分析出作業的瓶頸在哪，並給出優化建議。

GIAC全球互聯網架構大會深圳站將於2019年6月舉行，屆時有更多Flink、Clickhouse等國內外大數據前沿、大數據平台相關演講。參加2019年GIAC深圳站，可以了解業界動態，和業界專家近距離接觸。

參加 GIAC，盤點2019年最新技術，目前購買8折優惠，多人購買有更多優惠。識別二維碼了解大會更多詳情。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！