1.背景 Apache Flink 和 Apache Storm 是當前業界廣泛使用的兩個分散式實時計算框架。其中 Apache Storm(以下簡稱“Storm”)在美團點評實時計算業務中已有較為成熟的運用(可參考 Storm 的 可靠性保證測試),有管理平臺、常用 API 和相應的文檔,大量實時 ...
1.背景
Apache Flink 和 Apache Storm 是當前業界廣泛使用的兩個分散式實時計算框架。其中 Apache Storm(以下簡稱“Storm”)在美團點評實時計算業務中已有較為成熟的運用(可參考 Storm 的 可靠性保證測試),有管理平臺、常用 API 和相應的文檔,大量實時作業基於 Storm 構建。而 Apache Flink(以下簡稱“Flink”)在近期倍受關註,具有高吞吐、低延遲、高可靠和精確計算等 特性,對事件視窗有很好的支持,目前在美團點評實時計算業務中也已有一定應用。
為深入熟悉瞭解 Flink 框架,驗證其穩定性和可靠性,評估其實時處理性能,識別該體系中的 缺點,找到其性能瓶頸併進行優化,給用戶提供最適合的實時計算引擎,我們以實踐經驗豐富 的 Storm 框架作為對照,進行了一系列實驗測試 Flink 框架的性能,計算 Flink 作為確保“至 少一次”和“恰好一次”語義的實時計算框架時對資源的消耗,為實時計算平臺資源規劃、框 架選擇、性能調優等決策及 Flink 平臺的建設提出建議並提供數據支持,為後續的 SLA 建設提供一定參考。
Flink 與 Storm 兩個框架對比:
流計算框架Flink與Storm 的性能對比
Storm | Flink | |
---|---|---|
狀態管理 | 無狀態,需用戶自行進行狀態管理 | 有狀態 |
視窗支持 | 對事件視窗支持較弱,緩存整個視窗的所有 數據,視窗結束時一起計算 | 視窗支持較為完善,自帶一些視窗聚合方法,並 且會自動管理視窗狀態。 |
消息投遞 | At Most Once At Least Once | At Most Once At Least Once Exactly Once |
容錯方式 | ACK機制:對每個消息進行全鏈路跟蹤,失敗 或超時進行重發。 | 檢查點機制:通過分散式一致性快照機制,對數 據流和運算元狀態進行保存。在發生錯誤時,使系 統能夠進行回滾。 |
應用現狀 | 在美團點評實時計算業務中已有較為成熟的 運用,有管理平臺、常用 API 和相應的文檔, 大量實時作業基於 Storm 構建。 | 在美團點評實時計算業務中已有一定應用,但 是管理平臺、API 及文檔等仍需進一步完善。 |
2.測試目標
評估不同場景、不同數據壓力下 Flink 和 Storm 兩個實時計算框架目前的性能表現,獲取其詳 細性能數據並找到處理性能的極限;瞭解不同配置對 Flink 性能影響的程度,分析各種配置的 適用場景,從而得出調優建議。
2.1 測試場景
“輸入-輸出”簡單處理場景
通過對“輸入-輸出”這樣簡單處理邏輯場景的測試,儘可能減少其它因素的干擾,反映兩個框 架本身的性能。
同時測算框架處理能力的極限,處理更加複雜的邏輯的性能不會比純粹“輸入-輸出”更高。
用戶作業耗時較長的場景
如果用戶的處理邏輯較為複雜,或是訪問了資料庫等外部組件,其執行時間會增大,作業的性 能會受到影響。因此,我們測試了用戶作業耗時較長的場景下兩個框架的調度性能。
視窗統計場景
實時計算中常有對時間視窗或計數視窗進行統計的需求,例如一天中每五分鐘的訪問量,每 100 個訂單中有多少個使用了優惠等。Flink 在視窗支持上的功能比 Storm 更加強大,API 更 加完善,但是我們同時也想瞭解在視窗統計這個常用場景下兩個框架的性能。
精確計算場景(即消息投遞語義為“恰好一次”)
Storm 僅能保證“至多一次” (At Most Once) 和“至少一次” (At Least Once) 的消息投遞語義, 即可能存在重覆發送的情況。有很多業務場景對數據的精確性要求較高,希望消息投遞不重不 漏。Flink 支持“恰好一次” (Exactly Once) 的語義,但是在限定的資源條件下,更加嚴格的精 確度要求可能帶來更高的代價,從而影響性能。因此,我們測試了在不同消息投遞語義下兩個 框架的性能,希望為精確計算場景的資源規劃提供數據參考。
2.2 性能指標
- 吞吐量(Throughput)
- 單位時間內由計算框架成功地傳送數據的數量,本次測試吞吐量的單位為:條/秒。
- 反映了系統的負載能力,在相應的資源條件下,單位時間內系統能處理多少數據。 •
- 吞吐量常用於資源規劃,同時也用於協助分析系統性能瓶頸,從而進行相應的資源調整以 保證系統能達到用戶所要求的處理能力。假設商家每小時能做二十份午餐(吞吐量 20 份/ 小時),一個外賣小哥每小時只能送兩份(吞吐量 2 份/小時),這個系統的瓶頸就在小哥配 送這個環節,可以給該商家安排十個外賣小哥配送。
- 延遲(Latency)
- 數據從進入系統到流出系統所用的時間,本次測試延遲的單位為:毫秒。
- 反映了系統處理的實時性。
- 金融交易分析等大量實時計算業務對延遲有較高要求,延遲越低,數據實時性越強。
- 假設商家做一份午餐需要 5 分鐘,小哥配送需要 25 分鐘,這個流程中用戶感受到了 30 分鐘的延遲。如果更換配送方案後延遲變成了 60 分鐘,等送到了飯菜都涼了,這個新的方案就是無法接受的。
3.測試環境
為 Storm 和 Flink 分別搭建由 1 台主節點和 2 台從節點構成的 Standalone 集群進行本次測試。其中為了觀察 Flink 在實際生產環境中的性能,對於部分測內容也進行了 on Yarn 環境的測試。
3.1 集群參數
參數項 | 參數值 |
---|---|
CPU | QEMU Virtual CPU version 1.1.2 2.6GHz |
Core | 8 |
Memory | 16GB |
Disk | 500G |
OS | CentOS release 6.5 (Final) |
3.2 框架參數
參數項 | Storm 配置 | Flink 配置 |
---|---|---|
Version | Storm 1.1.0-mt002 | Flink 1.3.0 |
Master Memory | 2600M | 2600M |
Slave Memory | 1600M * 16 | 12800M * 2 |
Parallelism | 2 supervisor 16 worker |
2 Task Manager 16 Task slots |
4.測試方法
4.1 測試流程
數據生產
Data Generator 按特定速率生成數據,帶上自增的 id 和 eventTime 時間戳寫入 Kafka 的一個 Topic(Topic Data)。
數據處理
Storm Task 和 Flink Task (每個測試用例不同)從 Kafka Topic Data 相同的 Offset 開始消費, 並將結果及相應 inTime、outTime 時間戳分別寫入兩個 Topic(Topic Storm 和 Topic Flink)中。
指標統計
Metrics Collector 按 outTime 的時間視窗從這兩個 Topic 中統計測試指標,每五分鐘將相應的 指標寫入 MySQL 表中。
Metrics Collector 按 outTime 取五分鐘的滾動時間視窗,計算五分鐘的平均吞吐(輸出數據的 條數)、五分鐘內的延遲(outTime - eventTime 或 outTime - inTime)的中位數及 99 線等指標, 寫入 MySQL 相應的數據表中。最後對 MySQL 表中的吞吐計算均值,延遲中位數及延遲 99 線 選取中位數,繪製圖像並分析。
4.2 預設參數
- Storm 和 Flink 預設均為At Least Once語義。
Storm 開啟 ACK,ACKer 數量為 1。
Flink 的 Checkpoint 時間間隔為 30 秒,預設 StateBackend 為 Memory。
保證 Kafka 不是性能瓶頸,儘可能排除 Kafka 對測試結果的影響。
測試延遲時數據生產速率小於數據處理能力,假設數據被寫入 Kafka 後立刻被讀取,即 eventTime 等於數據進入系統的時間。
測試吞吐量時從 Kafka Topic 的最舊開始讀取,假設該 Topic 中的測試數據量充足。
4.3 測試用例
Identity
Identity 用例主要模擬“輸入-輸出”簡單處理場景,反映兩個框架本身的性能。
輸入數據為“msgId, eventTime”,其中 eventTime 視為數據生成時間。單條輸入數據約 20 B。
進入作業處理流程時記錄 inTime,作業處理完成後(準備輸出時)記錄 outTime。
作業從 Kafka Topic Data 中讀取數據後,在字元串末尾追加時間戳,然後直接輸出到 Kafka。
輸出數據為“msgId, eventTime, inTime, outTime”。單條輸出數據約 50 B。
Sleep
- Sleep 用例主要模擬用戶作業耗時較長的場景,反映複雜用戶邏輯對框架差異的削弱,比較 兩個框架的調度性能。
- 輸入數據和輸出數據均與 Identity 相同。
- 讀入數據後,等待一定時長(1 ms)後在字元串末尾追加時間戳後輸出
Windowed Word Count
- Windowed Word Count 用例主要模擬視窗統計場景,反映兩個框架在進行視窗統計時性能 的差異。
- 此外,還用其進行了精確計算場景的測試,反映 Flink 恰好一次投遞的性能。
- 輸入為 JSON 格式,包含 msgId、eventTime 和一個由若幹單片語成的句子,單詞之間由空 格分隔。單條輸入數據約 150 B。
- 讀入數據後解析 JSON,然後將句子分割為相應單詞,帶 eventTime 和 inTime 時間戳發給 CountWindow 進行單詞計數,同時記錄一個視窗中最大最小的 eventTime 和 inTime,最後 帶 outTime 時間戳輸出到 Kafka 相應的 Topic。
- Spout/Source 及 OutputBolt/Output/Sink 併發度恆為 1,增大併發度時僅增大 JSONParser、 CountWindow 的併發度。
- 由於 Storm 對 window 的支持較弱,CountWindow 使用一個 HashMap 手動實現,Flink 用了原生的 CountWindow 和相應的 Reduce 函數。
5.測試結果
5.1 Identity 單線程吞吐量
- 上圖中藍色柱形為單線程 Storm 作業的吞吐,橙色柱形為單線程 Flink 作業的吞吐。
- Identity 邏輯下,Storm 單線程吞吐為8.7萬條/秒,Flink 單線程吞吐可達35萬條/秒。
- 當 Kafka Data 的 Partition 數為 1 時,Flink 的吞吐約為 Storm 的 3.2 倍;當其 Partition 數為 8 時,Flink 的吞吐約為 Storm 的 4.6 倍。
- 由此可以看出,Flink 吞吐約為 Storm 的 3-5 倍。
5.2 Identity 單線程作業延遲
- 採用 outTime - eventTime 作為延遲,圖中藍色折線為 Storm,橙色折線為 Flink。虛線為 99 線,實線為中位數。
- 從圖中可以看出隨著數據量逐漸增大,Identity 的延遲逐漸增大。其中 99 線的增大速度比中位數快,Storm 的 增大速度比 Flink 快。
- 其中 QPS 在 80000 以上的測試數據超過了 Storm 單線程的吞吐能力,無法對 Storm 進 行測試,只有 Flink 的曲線。
- 對比折線最右端的數據可以看出,Storm QPS 接近吞吐時延遲中位數約 100 毫秒,99 線約 700 毫秒,Flink 中位數約 50 毫秒,99 線約 300 毫秒。Flink 在滿吞吐時的延遲約為 Storm 的一半。
5.3 Sleep 吞吐量
- 從圖中可以看出,Sleep 1 毫秒時,Storm 和 Flink 單線程的吞吐均在 900 條/秒左右,且隨著併發增大基本呈線性增大。
- 對比藍色和橙色的柱形可以發現,此時兩個框架的吞吐能力基本一致。
5.4 Sleep 單線程作業延遲(中位數)
- 依然採用 outTime - eventTime 作為延遲,從圖中可以看出,Sleep 1 毫秒時,Flink 的延遲仍低於 Storm。
5.5 Windowed Word Count 單線程吞吐量
- 單線程執行大小為 10 的計數視窗,吞吐量統計如圖。
- 從圖中可以看出,Storm 吞吐約為 1.2 萬條/秒,Flink Standalone 約為 4.3 萬條/秒。Flink 吞吐依然為 Storm 的 3 倍以上。
5.6 Windowed Word Count Flink At Least Once 與 Exactly Once 吞吐量對比
由於同一運算元的多個並行任務處理速度可能不同,在上游運算元中不同快照里的內容,經過中間並行運算元的處理,到達下游運算元時可能被計入同一個快照中。這樣一來,這部分數據會 被重覆處理。因此,Flink 在 Exactly Once 語義下需要進行對齊,即當前最早的快照中所有 數據處理完之前,屬於下一個快照的數據不進行處理,而是在緩存區等待。當前測試用例 中,在 JSON Parser 和 CountWindow、CountWindow 和 Output 之間均需要進行對齊,有 一定消耗。為體現出對齊場景,Source/Output/Sink 併發度的併發度仍為 1,提高了 JSONParser/CountWindow 的併發度。具體流程細節參見前文 Windowed Word Count 流程圖。
上圖中橙色柱形為 At Least Once 的吞吐量,黃色柱形為 Exactly Once 的吞吐量。對比兩者可以看出,在當前併發條件下,Exactly Once 的吞吐較 At Least Once 而言下降了 6.3%
5.7 Windowed Word Count Storm At Least Once 與 At Most Once 吞吐量對比
- Storm 將 ACKer 數量設置為零後,每條消息在發送時就自動 ACK,不再等待 Bolt 的 ACK, 也不再重發消息,為 At Most Once 語義。
- 上圖中藍色柱形為 At Least Once 的吞吐量,淺藍色柱形為 At Most Once 的吞吐量。對比兩者可以看出,在當前併發條件下,At Most Once 語義下的吞吐較 At Least Once 而言提高了 16.8%
5.8 Windowed Word Count 單線程作業延遲
Identity 和 Sleep 觀測的都是 outTime - eventTime,因為作業處理時間較短或 Thread.sleep() 精度不高,outTime - inTime 為零或沒有比較意義;Windowed Word Count 中可以有效測得 outTime - inTime 的數值,將其與 outTime - eventTime 畫在同一張圖上,其中 outTime - eventTime 為虛線,outTime - InTime 為實線。 • 觀察橙色的兩條折線可以發現,Flink 用兩種方式統計的延遲都維持在較低水平;觀察兩條 藍色的曲線可以發現,Storm 的 outTime - inTime 較低,outTime - eventTime 一直較高,即 inTime 和 eventTime 之間的差值一直較大,可能與 Storm 和 Flink 的數據讀入方式有關。
藍色折線表明 Storm 的延遲隨數據量的增大而增大,而橙色折線表明 Flink 的延遲隨著數 據量的增大而減小(此處未測至 Flink 吞吐量,接近吞吐時 Flink 延遲依然會上升)。 • 即使僅關註 outTime - inTime(即圖中實線部分),依然可以發現,當 QPS 逐漸增大的時候, Flink 在延遲上的優勢開始體現出來。
5.9 Windowed Word Count Flink At Least Once 與 Exactly Once 延遲對比
- 圖中黃色為 99 線,橙色為中位數,虛線為 At Least Once,實線為 Exactly Once。圖中相應 顏色的虛實曲線都基本重合,可以看出 Flink Exactly Once 的延遲中位數曲線與 At Least Once 基本貼合,在延遲上性能沒有太大差異。
5.10 Windowed Word Count Storm At Least Once 與 At Most Once 延遲對比
- 圖中藍色為 99 線,淺藍色為中位數,虛線為 At Least Once,實線為 At Most Once。QPS 在 4000 及以前的時候,虛線實線基本重合;QPS 在 6000 時兩者已有差異,虛線略高;QPS 接近 8000 時,已超過 At Least Once 語義下 Storm 的吞吐,因此只有實線上的點。 • 可以看出,QPS 較低時 Storm At Most Once 與 At Least Once 的延遲觀察不到差異,隨著 QPS 增大差異開始增大,At Most Once 的延遲較低。
5.11 Windowed Word Count Flink 不同 StateBackends 吞吐量對比
- Flink 支持 Standalone 和 on Yarn 的集群部署模式,同時支持 Memory、FileSystem、RocksDB 三種狀態存儲後端(StateBackends)。由於線上作業需要,測試了這三種 StateBackends 在 兩種集群部署模式上的性能差異。其中,Standalone 時的存儲路徑為 JobManager 上的一 個文件目錄,on Yarn 時存儲路徑為 HDFS 上一個文件目錄。
- 對比三組柱形可以發現,使用 FileSystem 和 Memory 的吞吐差異不大,使用 RocksDB 的 吞吐僅其餘兩者的十分之一左右。
- 對比兩種顏色可以發現,Standalone 和 on Yarn 的總體差異不大,使用 FileSystem 和 Memory 時 on Yarn 模式下吞吐稍高,使用 RocksDB 時 Standalone 模式下的吞吐稍高。
5.12 Windowed Word Count Flink 不同 StateBackends 延遲對比
使用 FileSystem 和 Memory 作為 Backends 時,延遲基本一致且較低。
使用 RocksDB 作為 Backends 時,延遲稍高,且由於吞吐較低,在達到吞吐瓶頸前的延遲陡增。其中 on Yarn 模式下吞吐更低,接近吞吐時的延遲更高。
6.結論及建議
6.1 框架本身性能
由 5.1、5.5 的測試結果可以看出,Storm 單線程吞吐約為 8.7 萬條/秒,Flink 單線程吞吐 可達 35 萬條/秒。Flink 吞吐約為 Storm 的 3-5 倍。
由 5.2、5.8 的測試結果可以看出,Storm QPS 接近吞吐時延遲(含 Kafka 讀寫時間)中位 數約 100 毫秒,99 線約 700 毫秒,Flink 中位數約 50 毫秒,99 線約 300 毫秒。Flink 在 滿吞吐時的延遲約為 Storm 的一半,且隨著 QPS 逐漸增大,Flink 在延遲上的優勢開始體現出來。
綜上可得,Flink 框架本身性能優於 Storm。
6.2 複雜用戶邏輯對框架差異的削弱
對比 5.1 和 5.3、5.2 和 5.4 的測試結果可以發現,單個 Bolt Sleep 時長達到 1 毫秒時, Flink 的延遲仍低於 Storm,但吞吐優勢已基本無法體現。
因此,用戶邏輯越複雜,本身耗時越長,針對該邏輯的測試體現出來的框架的差異越小。
6.3 不同消息投遞語義的差異
- 由 5.6、5.7、5.9、5.10 的測試結果可以看出,Flink Exactly Once 的吞吐較 At Least Once 而 言下降 6.3%,延遲差異不大;Storm At Most Once 語義下的吞吐較 At Least Once 提升 16.8%,延遲稍有下降。
- 由於 Storm 會對每條消息進行 ACK,Flink 是基於一批消息做的檢查點,不同的實現原理導 致兩者在 At Least Once 語義的花費差異較大,從而影響了性能。而 Flink 實現 Exactly Once 語義僅增加了對齊操作,因此在運算元併發量不大、沒有出現慢節點的情況下對 Flink 性能的 影響不大。Storm At Most Once 語義下的性能仍然低於 Flink。
6.4 Flink 狀態存儲後端選擇
• Flink 提供了記憶體、文件系統、RocksDB 三種 StateBackends,結合 5.11、5.12 的測試結果, 三者的對比如下:
StateBackend 過程狀態存儲 檢查點存儲 吞吐 推薦使用場景 Memory TM Memory JM Memory 高(3-5 倍 Storm) 調試、無狀態或對數據是否 丟失重覆無要求 FileSystem TM Memory FS/HDFS 高(3-5 倍 Storm) 普通狀態、視窗、KV 結構 (建議作為預設 Backend)
RocksDB RocksDB on TM FS/HDFS 低(0.3-0.5 倍 Storm) 超大狀態、超長視窗、大型 KV 結構
6.5 推薦使用 Flink 的場景
綜合上述測試結果,以下實時計算場景建議考慮使用 Flink 框架進行計算:
要求消息投遞語義為Exactly Once的場景;
數據量較大,要求高吞吐低延遲的場景;
需要進行狀態管理或視窗統計的場景。
7.展望
本次測試中尚有一些內容沒有進行更加深入的測試,有待後續測試補充。例如:
Exactly Once 在併發量增大的時候是否吞吐會明顯下降?
用戶耗時到 1ms 時框架的差異已經不再明顯(Thread.sleep() 的精度只能到毫秒),用 戶耗時在什麼範圍內 Flink 的優勢依然能體現出來?
- 本次測試僅觀察了吞吐量和延遲兩項指標,對於系統的可靠性、可擴展性等重要的性能指 標沒有在統計數據層面進行關註,有待後續補充。
- Flink 使用 RocksDBStateBackend 時的吞吐較低,有待進一步探索和優化。
關於 Flink 的更高級 API,如 Table API & SQL 及 CEP 等,需要進一步瞭解和完善。
8.參考內容
分散式流處理框架——功能對比和性能評估
intel-hadoop/HiBench: HiBench is a big data benchmark suite
Yahoo的流計算引擎基準測試
Extending the Yahoo! Streaming Benchmark
本文選自《不僅僅是流計算 Apache Flink實踐》
更多Flink博文:
更多Flink原理知識:
更多實時計算,Flink,Kafka等相關技術博文,歡迎關註實時流式計算: