分布式流處理系統(tǒng)的容錯性能基準(zhǔn)測試

蔣程　王曉桐　張蓉

摘? 要：隨著對數(shù)據(jù)處理的實時性要求越來越高，分布式流處理系統(tǒng)應(yīng)運而生。但是在分布式的集群規(guī)模下，各種軟硬件原因?qū)е碌墓收虾茈y避免的。現(xiàn)有的相關(guān)基準(zhǔn)測試主要關(guān)注于分布式流處理系統(tǒng)的處理性能，很少對該類系統(tǒng)處理故障的容錯性能進(jìn)行評測，以至于關(guān)鍵應(yīng)用在系統(tǒng)選型的時候特別艱難。針對分布式流處理系統(tǒng)的容錯性能，本文設(shè)計并實現(xiàn)了一套靈活的基準(zhǔn)測試框架。最后，本文在開源數(shù)據(jù)流處理系統(tǒng)Apache Storm和Apache Flink進(jìn)行了容錯性能的基準(zhǔn)測試，驗證定義的測試基準(zhǔn)的正確性和有效性，實驗結(jié)果也表明Flink的容錯性能相對較好。

關(guān)鍵詞：分布式系統(tǒng);流處理;容錯性能;基準(zhǔn)測試

中圖分類號：TP302.8? ? ?文獻(xiàn)標(biāo)識碼：A

Benchmarking for Fault-tolerant Performance in Distributed

Stream Processing Systems

JIANG Cheng，WANG Xiaotong，ZHANG Rong

（School of Data Science and Engineering，East China Normal University，Shanghai 200062，China）

Abstract：With the increasing real-time requirements for data processing，distributed stream processing systems have emerged.However，under the distributed cluster scale，failures caused by various hardware and software problems are inevitable.The existing related benchmarking mainly focus on the performance of the distributed stream processing system during failure-free time，while rarely evaluating the fault-tolerant performance of the system for handling faults.As a result，it is particularly difficult to select a system for mission-critical applications.This paper designs and implements a flexible benchmarking framework tailored for fault-tolerant performance.Finally，benchmarking the fault-tolerant performance of Apache Storm and Apache Flink verifies the correctness and effectiveness of the benchmark defined in this paper.Experimental results show that fault-tolerant performance of Flink outperforms that of Storm.

Keywords：distributed system;stream processing;fault-tolerant performance;benchmarking

1? ?引言（Introduction）

分布式流計算系統(tǒng)[1]（Distributed Stream Processing Systems，DSPS）是對大規(guī)模流數(shù)據(jù)進(jìn)行實時處理的系統(tǒng)，主流的開源系統(tǒng)有：Apache Flink[2]、Apache Storm[3]、Apache Spark Streaming[4]等。流計算的常見應(yīng)用場景有：電商的商品推薦，IoT設(shè)備的監(jiān)控預(yù)警，銀行的金融欺詐檢測等。流計算應(yīng)用具有以下特征：①高性能：系統(tǒng)需要對流入的數(shù)據(jù)進(jìn)行實時處理，延遲一般在毫秒級別。而且由于數(shù)據(jù)的不斷流入，計算需要支持很高的吞吐（例如，推特每天處理5億推文，F(xiàn)acebook有14.5億活躍用戶）;②容錯性：因為流數(shù)據(jù)的無限性，系統(tǒng)的運行需要支持7×24小時服務(wù)。在大規(guī)模分布式計算中像節(jié)點故障，網(wǎng)絡(luò)錯誤等故障經(jīng)常發(fā)生。流處理系統(tǒng)需要使用容錯機(jī)制來應(yīng)對故障的發(fā)生。特別是對于金融領(lǐng)域的關(guān)鍵應(yīng)用而言，快速的故障恢復(fù)和計算結(jié)果的正確性保證尤其重要，否則將會導(dǎo)致嚴(yán)重的財力損失。本文提出了一套針對分布式流處理系統(tǒng)容錯性能的基準(zhǔn)測試框架。

2? ?相關(guān)工作（Related work）

Linear Road Benchmark[5]最早提出了評測流數(shù)據(jù)管理系統(tǒng)（Stream Data Management Systems，SDMS）[6]的處理能力。它模擬了高速公路管理系統(tǒng)，該系統(tǒng)能通過實時收集處理公路上汽車傳來的位置數(shù)據(jù)，提供收費、事故檢測和警報等功能。它用于評測Aurora[7]和關(guān)系型數(shù)據(jù)庫管理系統(tǒng)能滿足該公路系統(tǒng)的最大處理吞吐量。StreamBench[8]針對DSPS設(shè)計了七個微型（micro）工作負(fù)載和四種工作負(fù)載套件，設(shè)計了兩種真實的應(yīng)用場景：實時網(wǎng)站日志處理和網(wǎng)絡(luò)流量監(jiān)控。它測試了Storm和Spark Streaming的處理性能、穩(wěn)定性和容錯性能。但在容錯性能的評測方面，StreamBench僅僅比較有無故障發(fā)生時的吞吐和延遲。Yahoo！ Streaming Benchmark[9]模擬了廣告分析應(yīng)用，測試了Flink、Storm和Spark Streaming的延遲和吞吐。RIoTBench[10]設(shè)計了IoT應(yīng)用場景下的相關(guān)負(fù)載，含有27種IoT相關(guān)的微型負(fù)載和四種由微型負(fù)載合成的應(yīng)用負(fù)載，使用真實的IoT數(shù)據(jù)集評測了Storm的處理性能。Jeyhun等人[11]提出了針對含有連接和窗口等復(fù)雜操作的延遲計算方法，定義了系統(tǒng)的最大可持續(xù)吞吐量。它模擬了在線視頻游戲應(yīng)用的相關(guān)工作負(fù)載，評測了Flink、Storm和Spark Streaming三個系統(tǒng)的處理性能。Steffen等人[12]通過廣告分析、公路管理和出租車業(yè)務(wù)查詢這三種工作負(fù)載的測試，對比分析了Flink、Storm和Spark Streaming的性能瓶頸。它提出當(dāng)前的分布式流處理系統(tǒng)存在沒有充分利用硬件資源的問題，并提出了優(yōu)化的設(shè)計方案。

現(xiàn)有的DSPS基準(zhǔn)測試相關(guān)信息統(tǒng)計如表1所示。目前大家關(guān)注的還是在系統(tǒng)無錯情況下DSPS的處理性能，而沒有對容錯機(jī)制和性能影響做深度調(diào)查和研究。StreamBench雖然涉及了容錯度量，但它只是通過性能指標(biāo)的變化很粗略地估計故障對性能帶來的影響。本文第一次以評測DSPS的容錯機(jī)制作為研究對象，定義和實現(xiàn)了考察這些容錯機(jī)制關(guān)鍵因素的benchmark和測試框架，并定義了評測故障恢復(fù)機(jī)制優(yōu)劣的性能指標(biāo)。該測試框架可以有效地評測不同的容錯技術(shù)給系統(tǒng)性能帶來的影響，為不同應(yīng)用場景下流處理系統(tǒng)的選擇提供依據(jù)和參考。

3? ?容錯機(jī)制（Fault-tolerant mechanism）

本章節(jié)將介紹本文評測的兩個最典型分布式流處理系統(tǒng)的容錯機(jī)制。

3.1? ?Apache Flink

Flink根據(jù)分布式快照算法Chandy-Lamport Algorithm[13]設(shè)計出了分布式輕量級異步快照機(jī)制。Flink會定期地發(fā)送一個柵欄標(biāo)記到輸入的數(shù)據(jù)流中，從而把源頭的數(shù)據(jù)流按段切割成版本遞增的快照。當(dāng)接收到所有輸入流中的柵欄標(biāo)記后，算子會對當(dāng)前版本的計算狀態(tài)進(jìn)行快照操作，把狀態(tài)持久化存儲到HDFS[14]等可靠的分布式存儲系統(tǒng)。一旦所有的算子都確認(rèn)完成了快照操作，F(xiàn)link會記錄當(dāng)前版本的全局一致快照已經(jīng)完成。在恢復(fù)期間，F(xiàn)link首先會重新部署整個計算拓?fù)?接著每個算子從分布式存儲系統(tǒng)中加載各自最近版本的檢查點快照;然后根據(jù)快照的版本，數(shù)據(jù)源需重發(fā)從最近檢查點時刻到故障發(fā)生時刻的數(shù)據(jù)，從而保證了Exactly-Once消息處理語義[15]。

3.2? ?Apache Storm

Storm的容錯機(jī)制由消息管理機(jī)制和快照機(jī)制共同完成，保證了At-Least-Once消息處理語義。消息管理機(jī)制指Storm會追蹤每條流入系統(tǒng)的數(shù)據(jù)，為其后續(xù)生成的子消息維護(hù)一個“消息樹”。當(dāng)且僅當(dāng)子消息都被成功處理，消息樹才會被判定為成功處理。否則，系統(tǒng)會重發(fā)對應(yīng)的輸入源消息?？煺諜C(jī)制指Storm會將算子的計算狀態(tài)進(jìn)行持久化保存。與Flink的容錯機(jī)制類似，Storm會定期向數(shù)據(jù)流中插入“快照事務(wù)”的消息;算子接收到快照事務(wù)消息后觸發(fā)準(zhǔn)備操作與提交操作：收到“準(zhǔn)備”事務(wù)消息時，算子將當(dāng)前版本的狀態(tài)臨時持久化;收到“提交”事務(wù)消息時，算子將當(dāng)前版本的狀態(tài)持久化，并刪除臨時狀態(tài)。在恢復(fù)期間，算子的狀態(tài)會根據(jù)故障發(fā)生時的快照事務(wù)狀態(tài)做出相應(yīng)的恢復(fù)。如果快照事務(wù)處于“正在準(zhǔn)備”狀態(tài)，由于部分算子并沒有臨時持久化準(zhǔn)備階段的狀態(tài)，則所有算子回滾至最近穩(wěn)定的快照版本;如果快照事務(wù)處于“正在提交”狀態(tài)，由于所有算子都已經(jīng)臨時持久化準(zhǔn)備階段的狀態(tài)，則所有算子繼續(xù)原來的計算任務(wù)。故障導(dǎo)致未完全處理的消息會因為消息超時或者算子主動發(fā)送失敗消息而標(biāo)記成失敗狀態(tài)，由消息管理機(jī)制負(fù)責(zé)重發(fā)。

4? ?基準(zhǔn)測試框架（Benchmarking framework）

本章節(jié)介紹本文的基準(zhǔn)評測設(shè)計，主要按包含的三個部分：容錯相關(guān)的度量定義，速率可控的數(shù)據(jù)實時生成，特征可控的負(fù)載設(shè)計。評測框架如圖1所示，包括如下四個部分：①數(shù)據(jù)生成器負(fù)責(zé)按給定速率實時生成流數(shù)據(jù)。該框架中的數(shù)據(jù)生成主要指控制數(shù)據(jù)流的產(chǎn)生流量和數(shù)據(jù)分布特征，從而改變DSPS的計算節(jié)點處理量。數(shù)據(jù)集包括兩類：一是下載的公共數(shù)據(jù)集，二是合成數(shù)據(jù)集。②消息隊列Kafka負(fù)責(zé)輸入數(shù)據(jù)和結(jié)果的存儲。Kafka作為本文的消息傳輸組件，不僅負(fù)責(zé)實時傳輸生成的數(shù)據(jù)至DSPS中進(jìn)行消費，還負(fù)責(zé)存儲計算產(chǎn)生的結(jié)果消息。③DSPS負(fù)責(zé)運行拓?fù)淙蝿?wù)。DSPS根據(jù)負(fù)載配置，如算子并行度、狀態(tài)大小等參數(shù)，在集群上生成并運行分布式測試工作負(fù)載。④度量收集器負(fù)責(zé)收集并統(tǒng)計度量指標(biāo)。度量收集器具有獲取集群的資源利用情況、獲取DSPS的實時吞吐和獲取延遲信息并進(jìn)行統(tǒng)計等功能。

4.1? ?度量定義

根據(jù)流計算和容錯機(jī)制的特性，我們設(shè)計三個相關(guān)度量：延遲、資源利用率、故障恢復(fù)時間。

延遲：本文采用的是事務(wù)時間的延遲。數(shù)據(jù)產(chǎn)生的時候，該數(shù)據(jù)會存儲產(chǎn)生時刻的時間戳，這個時間戳稱為生產(chǎn)時間。輸入DSPS系統(tǒng)的數(shù)據(jù)稱為原始數(shù)據(jù)。經(jīng)過DSPS運算，原始數(shù)據(jù)可能產(chǎn)生多個子數(shù)據(jù)，子數(shù)據(jù)的生產(chǎn)時間按照原始數(shù)據(jù)的生產(chǎn)時間不變。輸出時間定義為子數(shù)據(jù)經(jīng)過DSPS計算處理后時間，但是不包含結(jié)果傳輸時間。這為了防止由結(jié)果存儲組件的不合理配置，性能瓶頸等原因可能造成因傳輸導(dǎo)致延遲增大的問題。一條數(shù)據(jù)的生產(chǎn)時間和輸出時間差稱為這條數(shù)據(jù)的事務(wù)時間延遲，如圖2所示。

每個元組延遲計算公式：

指結(jié)果算子接收到該元組的時間，指數(shù)據(jù)生成器生成該元組的時間。如果一條元組經(jīng)過計算產(chǎn)生多個子元組，那么子元組的跟產(chǎn)生該元組的原始元組相同。本文指的延遲是所有元組延遲的平均值。

資源利用率：容錯機(jī)制對狀態(tài)的存儲，傳輸?shù)炔僮鲿o系統(tǒng)帶來額外的資源使用。本文關(guān)注于節(jié)點在任務(wù)運行時間段內(nèi)的平均CPU使用率。

故障恢復(fù)時間：本文通過軟件的方法實現(xiàn)在節(jié)點中隨機(jī)終止DSPS的運算進(jìn)程從而達(dá)到模擬故障的效果。故障發(fā)生時間定義為故障腳本的啟動時間。故障恢復(fù)時間可從宏觀和微觀的角度進(jìn)行定義。宏觀的角度指：從故障發(fā)生到系統(tǒng)的吞吐恢復(fù)到正常數(shù)值（無故障情況下）的時間;微觀的角度指：故障恢復(fù)時間可具體劃分為重載時間和重播時間。重載時間指從故障發(fā)生后到算子經(jīng)過重新部署并且從存儲系統(tǒng)中重載快照數(shù)據(jù)所花費的時間，重播時間指數(shù)據(jù)源重播到故障前消費的數(shù)據(jù)所花費的時間。從故障發(fā)生的時刻到系統(tǒng)恢復(fù)故障前狀態(tài)的時刻，這一時間段稱為故障恢復(fù)時間，如圖3所示。

根據(jù)上述定義，在Flink中，故障恢復(fù)時間從微觀角度按故障發(fā)生的時間到數(shù)據(jù)源重新處理到故障前的數(shù)據(jù)時間計算;而在Storm中，由于快照機(jī)制和消息重播機(jī)制分離，故障恢復(fù)時間只能從宏觀角度按故障發(fā)生的時間到數(shù)據(jù)源的吞吐恢復(fù)穩(wěn)定的時間計算。

4.2? ?數(shù)據(jù)集與工作負(fù)載

本章節(jié)抽象出數(shù)據(jù)流的數(shù)據(jù)特征和有狀態(tài)負(fù)載的特征，通過調(diào)控特征參數(shù)，可以模擬并控制系統(tǒng)工作負(fù)載。

4.2.1? ?輸入數(shù)據(jù)流特征

數(shù)據(jù)流數(shù)據(jù)本身有三個可調(diào)控的特征參數(shù)。

輸入速率：為了使系統(tǒng)運行在穩(wěn)定的狀態(tài)，本文控制一個穩(wěn)定并且合適的數(shù)據(jù)生產(chǎn)速率，防止系統(tǒng)負(fù)載過高進(jìn)入反壓狀態(tài)[16]。

數(shù)據(jù)傾斜度：數(shù)據(jù)傾斜是數(shù)據(jù)集中常見的特性。不均勻的數(shù)據(jù)分布將會導(dǎo)致大量數(shù)據(jù)集中在某些節(jié)點，造成節(jié)點的運算負(fù)荷不同。本文按Zipf定律生成數(shù)據(jù)傾斜的合成數(shù)據(jù)集。

輸入數(shù)據(jù)大?。簲?shù)據(jù)流具有無限性，但是根據(jù)實驗需求，可根據(jù)輸入的吞吐速率和運行時間修改原始數(shù)據(jù)集大小，計算公式如下：

其中，L是修改后的數(shù)據(jù)集總量，P是設(shè)定的輸入吞吐速率（條/秒），T是任務(wù)運行的時間（秒）。

本文內(nèi)置數(shù)據(jù)集含有兩種：第一種是從古登堡計劃（Project Gutenberg）獲取的英文小說集;第二種是根據(jù)數(shù)據(jù)傾斜程度生成的合成數(shù)據(jù)集。數(shù)據(jù)生成器根據(jù)配置的輸入吞吐速率，實時從數(shù)據(jù)集中獲取數(shù)據(jù)并輸入到Kafka中，模擬生產(chǎn)環(huán)境中的實時數(shù)據(jù)生成。

4.2.2 工作負(fù)載設(shè)計

本文設(shè)計了兩類工作負(fù)載：①計算簡單、狀態(tài)大小可調(diào)控的Word Count負(fù)載[17];②狀態(tài)大小固定、計算密集程度可調(diào)控的圓周率計算負(fù)載。工作負(fù)載的特征如圖4所示。通過分析有狀態(tài)計算的特征，本文的工作負(fù)載設(shè)有兩個可調(diào)控的特征參數(shù)。

算子的狀態(tài)大?。河绊懘鎯蛡浞菁磎emory和磁盤

I/O。算子分為有狀態(tài)計算和無狀態(tài)計算。無狀態(tài)計算指不需要依賴歷史數(shù)據(jù)進(jìn)行計算的算子，如切分算子，只需對當(dāng)前數(shù)據(jù)進(jìn)行分詞操作。有狀態(tài)計算指當(dāng)前計算需要根據(jù)歷史數(shù)據(jù)進(jìn)行計算，如窗口算子，計算需要對到達(dá)窗口內(nèi)的所有數(shù)據(jù)或者計算的中間結(jié)果值等狀態(tài)進(jìn)行聚合計算或者更新。為了防止因為故障導(dǎo)致狀態(tài)的丟失，保證恢復(fù)后計算的準(zhǔn)確性，有狀態(tài)的算子需要對狀態(tài)進(jìn)行持久化存儲。本文使用全歷史計算而不使用窗口算子，因為窗口算子的狀態(tài)大小不可控。在DSPS中，連接操作需要使用窗口算子，本文也不使用。在窗口算子中，觸發(fā)checkpoint操作的時間點在窗口內(nèi)呈現(xiàn)無規(guī)則分布，這種現(xiàn)象導(dǎo)致每次實驗中checkpoint保存的狀態(tài)大小不一致，無法通過控制變量法研究狀態(tài)大小和checkpoint間隔對系統(tǒng)帶來的影響。本文在2號算子中根據(jù)配置參數(shù)進(jìn)行自定義大小的字符串類型狀態(tài)存儲，保證每次存儲的狀態(tài)大小一致，從而研究不同狀態(tài)大小和checkpoint間隔對系統(tǒng)的影響。

算子的計算密集程度：影響CPU。本文研究不同計算密集型的算子受checkpoint操作的影響。本文設(shè)計狀態(tài)大小固定的圓周率計算算子，通過傳入配置參數(shù)實現(xiàn)控制2號算子中格雷戈里-萊布尼茨級數(shù)的運算次數(shù)，以此來調(diào)控該算子的計算密集程度。格雷戈里-萊布尼茨級數(shù)的計算公式如下：

本文通過對抽象出的兩個特征的調(diào)控，可以模擬出其他工作負(fù)載的特征，比如含有窗口操作的負(fù)載需要對到達(dá)窗口內(nèi)的所有數(shù)據(jù)進(jìn)行保存，存儲狀態(tài)較大;含有連接操作的負(fù)載需要對多條輸入流進(jìn)行連接操作，連接算子的運算密集程度大，并且需要對多條流的數(shù)據(jù)都進(jìn)行保存，存儲狀態(tài)較大。

5? ?實驗（Evaluation）

5.1? ?實驗環(huán)境

本文的實驗在具有五個節(jié)點的集群上進(jìn)行，節(jié)點的操作系統(tǒng)版本是CentOS v.6.5。測試平臺為Apache Flink 1.7.0，Apache Storm 1.2.2。其中一個節(jié)點配置為24核Intel（R）Xeon（R）CPU E5-2620、頻率2.40GHz、內(nèi)存31GB，部署非計算組件，如HDFS、Zookeeper、Redis等服務(wù)。其余四個節(jié)點配置為8核Intel（R）Xeon（R）CPU E5606，頻率2.13GHz，內(nèi)存94GB，部署計算組件，如Flink中的Taskmanager，Storm中的Worker等計算進(jìn)程。計算組件和非計算組件的分開部署能提高度量指標(biāo)的準(zhǔn)確性，如資源利用率。節(jié)點之間通過千兆以太網(wǎng)連接。默認(rèn)的數(shù)據(jù)輸入吞吐速率為5000條/秒;數(shù)據(jù)集使用真實的英文小說集。

5.2? ?無故障性能評測

系統(tǒng)在未發(fā)生故障的時候，容錯機(jī)制對性能的影響源自周期性地進(jìn)行的快照操作，對計算產(chǎn)生的中間狀態(tài)進(jìn)行持久化存儲。Checkpoint操作的頻率和持久化的狀態(tài)大小均是影響系統(tǒng)性能的重要因素。本組實驗研究不同狀態(tài)大小和checkpoint間隔對延遲和CPU使用率的影響。

從圖5（a）和圖5（b）中可以看出，在Flink中，當(dāng)狀態(tài)大小保持一致時，checkpoint間隔越短，計算的延遲越大，系統(tǒng)的CPU消耗越多。因為越頻繁的checkpoint操作會導(dǎo)致系統(tǒng)花費更多的資源在狀態(tài)處理上，使得正常的計算暫停的時間越多;當(dāng)checkpoint間隔保持一致時，狀態(tài)越大，計算的延遲越大，系統(tǒng)的CPU消耗越多。因為狀態(tài)越大，每次對狀態(tài)的持久化操作所需時間更久，對性能造成的影響更大。在Storm平臺上1秒的checkpoint間隔過于頻繁并且較大的狀態(tài)會嚴(yán)重影響系統(tǒng)的性能，導(dǎo)致其無法正常運行。故checkpoint間隔始于30s。從圖5（c）和圖5（d）中可以看出，在Storm中，checkpoint間隔的影響和Flink稍有不同。Storm的容錯機(jī)制對系統(tǒng)處理的影響主要有兩種操作。第一種操作是對狀態(tài)的存儲，該操作造成的延遲受狀態(tài)大小的影響;第二種是對checkpoint間隔時間內(nèi)緩存的元組進(jìn)行消息管理操作，該操作造成的延遲受checkpoint間隔的影響。狀態(tài)較小時（0—5MB）第一種操作的延遲影響比第二種操作小。狀態(tài)較大時（5—15MB）第二種操作的延遲影響比第一種操作小。CPU使用率變化趨勢也是類似的情況，但是平衡點在10MB左右。關(guān)于狀態(tài)的影響，當(dāng)checkpoint間隔保持一致時，狀態(tài)與延遲和CPU使用率成線性關(guān)系。因為狀態(tài)越大，狀態(tài)持久化的操作所花費的時間越久，使得正常運算的延遲增大。

觀察Flink和Storm該組實驗，如狀態(tài)大小為10MB，checkpoint間隔為30s時，F(xiàn)link的延遲比Storm低，而且CPU使用率也更低。

Flink通過柵欄的對齊操作來保證Exactly-Once消息處理語義。本文通過生成合成數(shù)據(jù)來模擬數(shù)據(jù)傾斜程度的不同，圖6反應(yīng)不同柵欄到達(dá)時間對延遲的影響。本組實驗的研究參數(shù)：checkpoint模式為NCP（不開啟checkpoint）、CP+NA（開啟30秒間隔的checkpoint，但是不開啟對齊操作）和CP+A（開啟30秒間隔的checkpoint，并且開啟對齊操作）。

從圖6可以看出，在數(shù)據(jù)傾斜度較大時，對齊操作對延遲的影響很大。這是因為在數(shù)據(jù)傾斜程度均勻的時候，每個算子的多個輸入通道中的柵欄到達(dá)時間相近，對齊操作導(dǎo)致的堵塞時間較少，所以延遲無明顯增大。但是在數(shù)據(jù)傾斜程度較大的時候，因為一個算子含有多個輸入通道時，數(shù)據(jù)量較少的低負(fù)載通道中的柵欄會先到達(dá)。這時對齊操作會堵塞已到達(dá)柵欄的通道，等到數(shù)據(jù)量較多的高負(fù)載通道中的柵欄。不同通道的柵欄到達(dá)時間相差越大將會導(dǎo)致該算子的同步堵塞操作時間越長，最終延遲會因此增大。

圖7展示負(fù)載計算密集程度受checkpoint操作的影響。越頻繁的checkpoint操作會導(dǎo)致頻繁的線程調(diào)度，切換等問題，負(fù)載計算密集程度越高受其干擾的影響越大，最終導(dǎo)致計算的延遲增大。

5.3? ?故障實驗

本文模擬的故障實驗是進(jìn)程級別的故障。由于程序錯誤、計算資源限制等原因，某個計算進(jìn)程出錯的概率很大。本文在流計算任務(wù)穩(wěn)定運行一段時間后，使用軟件腳本隨機(jī)終止某個節(jié)點上的某個計算進(jìn)程，從而模擬計算進(jìn)程故障。本組實驗設(shè)置的固定條件與上組實驗相同。

Flink的故障恢復(fù)時間可以根據(jù)恢復(fù)階段來劃分成重載時間和重播時間。重載時間指從故障發(fā)生的時間到任務(wù)重新部署完成的時間。重播時間指任務(wù)部署完成到數(shù)據(jù)源重新消費到故障發(fā)生前數(shù)據(jù)時間。從圖8中可以看出，故障恢復(fù)時間中重載階段的耗時占比較大。因為系統(tǒng)探測到TaskManager故障的時間跟配置參數(shù)心跳超時時間成正相關(guān)關(guān)系。在默認(rèn)配置下，重載階段花費約45秒左右，總恢復(fù)時間在60秒之內(nèi)。從圖8（b）中可以看出，狀態(tài)的大小和重播時間成正相關(guān)關(guān)系。因為在Flink的恢復(fù)過程中，算子各自進(jìn)行恢復(fù)操作，狀態(tài)大導(dǎo)致算子的平均恢復(fù)時間大，數(shù)據(jù)源的重發(fā)速率受其影響。

Storm中的恢復(fù)時間采用從宏觀的角度評測，根據(jù)吞吐隨時間變化情況測量恢復(fù)時間，如圖9（a）所示。由于Storm的整體處理性能較低，本組實驗中數(shù)據(jù)輸入吞吐速率降為500條/秒，

從而保證Storm能正常故障恢復(fù)。Storm的故障恢復(fù)由消息管理機(jī)制與快照機(jī)制共同完成，二者相互獨立，且前者對性能的影響占主導(dǎo)因素。Storm恢復(fù)故障算子的時候不需要部署整個任務(wù)，只需重啟故障的計算進(jìn)程，這部分操作耗時約在10秒。但是消息重發(fā)階段需要等待消息超時后由消息管理機(jī)制負(fù)責(zé)重發(fā)。本實驗在保證實驗正常運行的情況下，研究不同checkpoint間隔對恢復(fù)時間的影響。從圖9（b）中可以看出，checkpoint間隔越大，恢復(fù)時間越長。因為checkpoint間隔影響了消息超時的時間，越長的checkpoint間隔導(dǎo)致失敗的消息被判定超時并且重發(fā)的所需時間越久，所以恢復(fù)時間越久。

對比Flink和Storm的故障實驗，即使在較高的輸入吞吐和較大的狀態(tài)下，F(xiàn)link的恢復(fù)時間更低，并且保證的語義更強(qiáng)，總體性能優(yōu)于Storm。

6? ?結(jié)論（Conclusion）

本文提出一種針對分布式流處理系統(tǒng)的容錯性能評測框架，使用真實和模擬的數(shù)據(jù)集，定義了影響容錯性能的負(fù)載特征以及容錯評估指標(biāo)，評測了Flink和Storm的容錯性能。在非故障期間，對容錯機(jī)制對系統(tǒng)的性能影響進(jìn)行了評測;在故障發(fā)生后，對系統(tǒng)的恢復(fù)時間進(jìn)行了評測。實驗結(jié)果表明，F(xiàn)link的容錯機(jī)制不僅保證了更高級的處理語義，而且對系統(tǒng)的性能影響較小，故障恢復(fù)也更快速。未來，我們將在幾方面開展工作：評測其他分布式流處理系統(tǒng);增加輸入流的相關(guān)特征控制，如動態(tài)變化的輸入速率、動態(tài)變化的skew分布，更真實地模擬生產(chǎn)環(huán)境;添加復(fù)雜工作負(fù)載;加入恢復(fù)準(zhǔn)確性等評測指標(biāo)。

參考文獻(xiàn)（References）

[1] Cherniack M，Balakrishnan H，Balazinska M，et al.Scalable Distributed Stream Processing[C].CIDR，2003，3：257-268.

[2] Carbone P，Katsifodimos A，Ewen S，et al.Apache flink：Stream and batch processing in a single engine[J].Bulletin of the IEEE Computer Society Technical Committee on Data Engineering，2015，36（4）：28-38.

[3] Toshniwal A，Taneja S，Shukla A，et al.Storm@ twitter[C].Proceedings of the 2014 ACM SIGMOD international conference on Management of data.ACM，2014：147-156.

[4] Zaharia M，Das T，Li H，et al.Discretized streams：Fault-tolerant streaming computation at scale[C].Proceedings of the twenty-fourth ACM symposium on operating systems principles.ACM，2013：423-438.

[5] Arasu A，Cherniack M，Galvez E，et al.Linear road：a stream data management benchmark[C].Proceedings of the Thirtieth international conference on Very large data bases-Volume 30.VLDB Endowment，2004：480-491.

[6] 金澈清，錢衛(wèi)寧，周傲英.流數(shù)據(jù)分析與管理綜述[J].軟件學(xué)報，2004（08）：1172-1181.

[7] Abadi D J，Carney D，?etintemel U，et al.Aurora：a new model and architecture for data stream management[J].the VLDB Journal，2003，12（2）：120-139.

[8] Lu R，Wu G，Xie B，et al.Stream bench：Towards benchmarking modern distributed stream computing frameworks[C].2014 IEEE/ACM 7th International Conference on Utility and Cloud Computing.IEEE，2014：69-78.

[9] Chintapalli S，Dagit D，Evans B，et al.Benchmarking streaming computation engines：Storm，flink and spark streaming[C].2016 IEEE international parallel and distributed processing symposium workshops （IPDPSW）.IEEE，2016：1789-1792.

[10] Shukla A，Chaturvedi S，Simmhan Y.Riotbench：a real-time iot benchmark for distributed stream processing platforms[J].arXiv preprint arXiv：1701.08530，2017.

[11] Karimov J，Rabl T，Katsifodimos A，et al.Benchmarking distributed stream processing engines[J].arXiv preprint arXiv：1802.08496，2018.

[12] Zeuch S，Monte B D，Karimov J，et al.Analyzing efficient stream processing on modern hardware[J].Proceedings of the VLDB Endowment，2019，12（5）：516-530.

[13] Mattern F.Efficient algorithms for distributed snapshots and global virtual time approximation[J].Journal of parallel and distributed computing，1993，18（4）：423-434.

[14] Shvachko K，Kuang H，Radia S，et al.The hadoop distributed file system[C].MSST，2010，10：1-10.

[15] Lopez M A，Lobato A G P，Duarte O C M B.A performance comparison of open-source stream processing platforms[C].2016 IEEE Global Communications Conference （GLOBECOM）.IEEE，2016：1-6.

[16] 熊安萍，朱恒偉，羅宇豪.Storm流式計算框架反壓機(jī)制研究[J].計算機(jī)工程與應(yīng)用，2018，54（1）：102-106.

[17] Ranger C，Raghuraman R，Penmetsa A，et al.Evaluating MapReduce for multi-core and multiprocessor systems[C].hpca.2007，7（3）：19.

作者簡介：

蔣? ?程（1995-），男，碩士生.研究領(lǐng)域：數(shù)據(jù)流基準(zhǔn)測試.

王曉桐（1994-），女，博士生.研究領(lǐng)域：分布式數(shù)據(jù)流處理，數(shù)據(jù)流基準(zhǔn)測試.

張? 蓉（1978-），女，博士，教授.研究領(lǐng)域：分布式數(shù)據(jù)管理.本文通訊作者.