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

蔣程　王曉桐　張蓉

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

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

中圖分類號(hào)：TP302.8? ? ?文獻(xiàn)標(biāo)識(shí)碼：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）

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

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

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

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

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

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

3.1? ?Apache Flink

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

3.2? ?Apache Storm

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

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

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

4.1? ?度量定義

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

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

每個(gè)元組延遲計(jì)算公式：

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

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

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

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

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ù)本身有三個(gè)可調(diào)控的特征參數(shù)。

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

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

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

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

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

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

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

算子的狀態(tài)大小：影響存儲(chǔ)和備份即memory和磁盤

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

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

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

5? ?實(shí)驗(yàn)（Evaluation）

5.1? ?實(shí)驗(yàn)環(huán)境

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

5.2? ?無(wú)故障性能評(píng)測(cè)

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

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

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

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

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

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

5.3? ?故障實(shí)驗(yàn)

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

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

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

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

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

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

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

作者簡(jiǎn)介：

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

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

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