史凌云,鄭瑩瑩,譚 勵,許利杰,王 偉,4,魏 峻,4

1(北京工商大學(xué) 計算機與信息工程學(xué)院,北京 100048)

2(中國科學(xué)院 軟件研究所,北京 100190)

3(中國科學(xué)院大學(xué),北京 100049)

4(計算機科學(xué)國家重點實驗室,北京 100190)

引言

隨著信息時代的到來,互聯(lián)網(wǎng)、物聯(lián)網(wǎng)、云計算等技術(shù)的飛速發(fā)展和廣泛應(yīng)用,數(shù)據(jù)在各個行業(yè)不斷地產(chǎn)生、積累并爆發(fā)式增長,已經(jīng)成為一種重要的生產(chǎn)因素,并滲透到每一個行業(yè)和業(yè)務(wù)職能領(lǐng)域[1].大數(shù)據(jù)被喻為“未來的新石油”[2],已經(jīng)成為社會各界關(guān)注的熱點,甚至成為各界爭奪的焦點,大數(shù)據(jù)時代已經(jīng)到來.相較于傳統(tǒng)的數(shù)據(jù),大數(shù)據(jù)具有數(shù)據(jù)規(guī)模大、數(shù)據(jù)類型多、數(shù)據(jù)處理速度快、數(shù)據(jù)價值密度低等特征,這些特征對大數(shù)據(jù)處理和應(yīng)用提出了更高的要求和更大的挑戰(zhàn).

目前對大數(shù)據(jù)處理的形式主要包括批式處理和流式處理[3].其中,批式處理是指對靜態(tài)有界數(shù)據(jù)的處理,對計算的實時性要求不高且應(yīng)用場景廣泛,具有代表性的批量大數(shù)據(jù)處理系統(tǒng)有Apache Hadoop[4]和Apache Spark[5]等.但是,隨著社交網(wǎng)絡(luò)、電子商務(wù)等技術(shù)的飛速發(fā)展和應(yīng)用,越來越多的應(yīng)用場景要求從海量的數(shù)據(jù)中及時獲取價值,并以很低的延遲來分析實時數(shù)據(jù).例如,以阿里巴巴為代表的電商平臺基于流式大數(shù)據(jù)處理系統(tǒng)實時統(tǒng)計和分析用戶行為,更新商品搜索引擎.因此,針對流式大數(shù)據(jù)的實時處理越來越流行,應(yīng)用場景也越來越重要.流式大數(shù)據(jù)處理系統(tǒng)的地位日漸凸顯,在業(yè)界也已經(jīng)有了非常廣泛的應(yīng)用,常見的有Apache Strom[6]、Apache Flink[7]、Apache Spark Streaming[8]等.

流式數(shù)據(jù)本身的實時性、難重復(fù)以及動態(tài)變化等特性,以及流式數(shù)據(jù)計算所需的數(shù)據(jù)無限性、計算有界性、計算實時性等特征,對流式大數(shù)據(jù)處理系統(tǒng)的性能和可靠性提出了更高的要求.流式大數(shù)據(jù)處理系統(tǒng)需要提供低延遲的數(shù)據(jù)處理,同時保證計算的正確性.然而,隨著集群規(guī)模的擴大,系統(tǒng)發(fā)生故障的概率也會增大,無法預(yù)知的錯誤可能會隨時出現(xiàn)在任意一個節(jié)點.一旦大數(shù)據(jù)處理系統(tǒng)出現(xiàn)問題,可能會產(chǎn)生不可挽回的損失.因此,針對流式大數(shù)據(jù)處理系統(tǒng)及其基準(zhǔn)測試框架的研究已經(jīng)成為了一個熱點問題.

現(xiàn)今已有多種流式大數(shù)據(jù)基準(zhǔn)測試框架,如Yahoo!Streaming Benchmark[9]和HiBench[10]等.但Yahoo!Streaming Benchmark 應(yīng)用場景單一,覆蓋程度較低;HiBench 僅能夠支持簡單流式數(shù)據(jù),本質(zhì)仍是一個批式大數(shù)據(jù)基準(zhǔn)測試框架.因此,現(xiàn)有流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試框架仍存在不足,比如應(yīng)用場景的設(shè)計較為簡單,評價指標(biāo)選取上較為單一,集中在吞吐量和延遲等.

針對現(xiàn)有流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試框架的不足和面臨的挑戰(zhàn),本文設(shè)計并實現(xiàn)了一個股票交易場景下的流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試框架.該框架包括流式數(shù)據(jù)生成方法、應(yīng)用場景構(gòu)造和評價指標(biāo)3 個部分,以Socket 作為流式數(shù)據(jù)源,選取真實的股票交易數(shù)據(jù)構(gòu)造了三個數(shù)據(jù)流,具體應(yīng)用覆蓋GroupBy 操作和Join 操作,并選取延遲、吞吐量、GC 時間和CPU利用率作為評價指標(biāo),構(gòu)建了一個實時計算與結(jié)構(gòu)化數(shù)據(jù)相結(jié)合的場景.此外,本文還針對數(shù)據(jù)輸入速率和執(zhí)行器內(nèi)核數(shù)量設(shè)計了兩個實驗,在Apache Spark Streaming 中對該框架進行實際的集群測試,對測試結(jié)果進行分析并得出結(jié)論,分析系統(tǒng)性能表現(xiàn),同時發(fā)現(xiàn)大數(shù)據(jù)處理系統(tǒng)存在的問題,并分析瓶頸所在,從而盡大可能地減少實際運行過程中可能出現(xiàn)的故障.本文的主要貢獻如下:

(1)總結(jié)了現(xiàn)有流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試框架特點及其不足.

(2)提出了一種流式數(shù)據(jù)生成方法,并使用多種測試指標(biāo)進行結(jié)果評測.

(3)設(shè)計并實現(xiàn)了一個基于股票交易場景的流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試框架.

(4)應(yīng)用流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試框架對Apache Spark Streaming 進行性能測試,發(fā)現(xiàn)并分析系統(tǒng)的不足.

1 相關(guān)工作

不同于批式大數(shù)據(jù)處理,流式大數(shù)據(jù)處理起步較晚,目前在業(yè)內(nèi)并沒有統(tǒng)一的基準(zhǔn)測試標(biāo)準(zhǔn).現(xiàn)有的流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試框架的相關(guān)研究如下.

Hesse 和Lorenz[11]從體系結(jié)構(gòu)等方面對比了Apache Storm、Flink、Spark Streaming 和Samza platforms.Gradvohl 等人[12]從系統(tǒng)容錯方面分析對比了Google Millwheel[13]、Yahoo Apache S4[14]、Spark Streaming 和Storm.然而,這兩篇文獻僅限于概念性討論,沒有實驗性的定量性能評估.Nabi 等人[15]提出了一個基準(zhǔn)測試,測量Apache Spark 和Apache Storm 的延遲和吞吐量,為流處理平臺的實驗比較創(chuàng)造了第一步.Dayarathna 和Suzumura[16]使用基準(zhǔn)測試比較了3 個流處理系統(tǒng)的吞吐量、CPU 和內(nèi)存消耗以及網(wǎng)絡(luò)使用情況.

此外,部分流式大數(shù)據(jù)處理系統(tǒng)的開發(fā)廠商選擇了自己認(rèn)為有代表性、能驗證系統(tǒng)功能性的應(yīng)用場景進行測試.如Yahoo!開發(fā)的分布式流計算平臺S4 選取了廣告點擊率計算(Click-Through Rate)進行性能驗證,以測試S4 處理流式數(shù)據(jù)的極限[2].Apache Storm 選取一個簡單的應(yīng)用,統(tǒng)計了不同應(yīng)用下參與的用戶數(shù),并測試其在故障下的表現(xiàn)[17].Apache Spark Streaming 僅對Grep、WordCount、TopKCount 這3 個常見應(yīng)用的吞吐量和故障恢復(fù)能力進行測試.應(yīng)用場景簡單以及流式計算特征的覆蓋率低使得這些測試框架無法全面的剖析流式大數(shù)據(jù)處理系統(tǒng)所面臨的性能及可靠性問題.

現(xiàn)有的針對多種流式大數(shù)據(jù)處理系統(tǒng)的基準(zhǔn)測試框架,其測試系統(tǒng)大多以Apache Storm、Apache Spark 和Apace Flink 為主.例如Lopez 等人[18]提出了一個針對Apache Storm、Apache Spark 和Apace Flink的基準(zhǔn)測試框架,測試了3 個系統(tǒng)在節(jié)點故障情況下的吞吐量.Karimov 等人[19]提出了一個分布式流處理引擎基準(zhǔn)測試框架,對Apache Storm、Apache Spark 和Apache Flink 的性能進行評估,并定義和測試流式大數(shù)據(jù)處理系統(tǒng)的可持續(xù)性能.由Yahoo!的一個團隊設(shè)計并實現(xiàn)的Yahoo! Streaming Benchmark 通過Kafka 和Redis 進行數(shù)據(jù)檢索和存儲,對Apache Storm、Apache Spark 和Apace Flink 進行實驗,測量了延遲和吞吐量[9].Perera 等人使用Yahoo! Streaming Benchmark 和Karamel[20]在云環(huán)境中提供Apache Spark 和Apache Flink 的可復(fù)制批處理和流基準(zhǔn)[21].但Yahoo! Streaming Benchmark 在應(yīng)用場景上覆蓋度較低,且只支持一個工作負(fù)載.

綜上所述,現(xiàn)有的流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試框架還存在各種不足,應(yīng)用用場景設(shè)計較為簡單,評價指標(biāo)選取上較為單一,集中在吞吐量和延遲.針對現(xiàn)有的流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試框架存在的不足,本文構(gòu)造了股票高頻交易場景,并選擇延遲、吞吐量、GC 時間和CPU 利用率作為評價指標(biāo).

2 流式大數(shù)據(jù)系統(tǒng)基準(zhǔn)測試框架設(shè)計與實現(xiàn)

本文基于流式大數(shù)據(jù)及其特征,設(shè)計并實現(xiàn)了一個流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試框架,包括流式數(shù)據(jù)生成方法、應(yīng)用場景構(gòu)造和評價指標(biāo).

通過流式數(shù)據(jù)生成方法,生成符合流式特征的股票交易數(shù)據(jù);通過應(yīng)用場景構(gòu)造,構(gòu)建一個股票交易場景用于系統(tǒng)測試;通過明確測試評價指標(biāo),收集并分析指標(biāo)數(shù)據(jù),從而分析系統(tǒng)性能并得出測試結(jié)論.基準(zhǔn)測試系統(tǒng)架構(gòu)圖如圖1 所示.

圖1 基準(zhǔn)測試系統(tǒng)架構(gòu)圖

2.1 流式數(shù)據(jù)生成方法

一般的批式大數(shù)據(jù)可以預(yù)先產(chǎn)生和存儲,使用方便.而由于流式計算的有界性和實時性等特點,數(shù)據(jù)生成需提供實時、高速的流式數(shù)據(jù),從而通過測試發(fā)現(xiàn)系統(tǒng)的瓶頸,保證基準(zhǔn)測試的有效性,這些都對流式大數(shù)據(jù)的生成方式提出了更為嚴(yán)格的要求.

Apache Spark Streaming 提供了兩種數(shù)據(jù)源,基礎(chǔ)數(shù)據(jù)源和高級數(shù)據(jù)源.基礎(chǔ)數(shù)據(jù)源是Streaming API 中直接提供的數(shù)據(jù)源,如socket 套接字、文件系統(tǒng)等.高級數(shù)據(jù)源是通過第三方類提供支持,如Kafka、Flume、Kinesis、Twitter 等.由于使用第三方工具生成數(shù)據(jù)本身會對性能有所影響[22],因此本文選用基礎(chǔ)的Socket傳輸方式作為Apache Spark Streaming 的數(shù)據(jù)源.Socket傳輸方式可通過程序控制向指定端口發(fā)送數(shù)據(jù),并可以通過調(diào)節(jié)參數(shù)改變數(shù)據(jù)的傳輸速度,以生成不同流速的流式大數(shù)據(jù).

針對股票高頻交易這一應(yīng)用場景,本文選擇了現(xiàn)實的股票交易數(shù)據(jù)作為數(shù)據(jù)源,主要涉及的數(shù)據(jù)類型為數(shù)值型和字符型,方便進行流式SQL 的計算.

2.2 應(yīng)用場景構(gòu)造

流式計算的應(yīng)用場景有很多,其中比較典型的是在金融銀行業(yè)的應(yīng)用.在金融銀行領(lǐng)域的日常運營過程中往往會產(chǎn)生大量的實時數(shù)據(jù),需要對這些海量數(shù)據(jù)進行實時分析處理以獲得其內(nèi)在價值,從而幫助金融銀行進行分析決策[23].其中本文選取的股票的高頻交易就是流式處理系統(tǒng)在金融銀行業(yè)的一個應(yīng)用.

(1)數(shù)據(jù)流構(gòu)造

實驗構(gòu)建了一個股票交易數(shù)據(jù)分析的場景,在滿足實際生產(chǎn)生活要求的同時,盡可能多地覆蓋流式計算特征.股票交易數(shù)據(jù)分析場景涉及3 個數(shù)據(jù)流:股票變化流、用戶交易流和用戶持倉流,如表1 所示.

表1 數(shù)據(jù)流設(shè)計

STOCK 是股票變化流,用于描述不同時間點股票的價格情況.其中,szcode 代表股票編號,eventTime 代表當(dāng)前時間,lastPrice 代表當(dāng)前價格.

TRANSACTION 是股票交易流,用于描述不同時間點股票交易情況.其中,szcode 代表交易的股票編號,userID 代表用戶編號,eventTime 代表交易時間,Turnover 代表成交量,Price 代表成交價格.

POSITION 是用戶持倉流,用于描述不同時間點用戶股票持倉情況.其中,userID 代表用戶編號,szcode代表該用戶持有的股票編號,lastPrice 代表上次成交價格,openInterest 代表持倉量.

(2)具體應(yīng)用構(gòu)造

本文主要實現(xiàn)了對GroupBy 和Join 應(yīng)用的覆蓋,設(shè)計如下:

1)GroupBy

GroupBy 是Apache Spark 中基本、常見的API,相當(dāng)于SQL 查詢中的groupby()函數(shù).實驗中實時獲取用戶交易流的數(shù)據(jù),并按照股票編號這一字段進行聚合,計算每只股票在一段時間內(nèi)的總成交額.

# SQL Query (GroupBy)

#實時計算n 分鐘內(nèi)每只股票的成交額.

SELECT szcode,SUM(Price*Turnover)

FROM TRANSACTION [Range n,Slide s]

GROUP BY szcode

2)Join

實驗中對股票變化流和用戶持倉流進行Join 操作,按照股票編號進行連接,實現(xiàn)對各個用戶持倉股票市值的實時計算.

# SQL Query (Join)

#每個用戶所持股票的市值(用戶持有量*當(dāng)前價格)

SELECT c.userID,SUM(lastPrice* openInterest)

FROM POSITION[Range n,Slide s] as p,STOCK[Range n,Slide s] as s,

ON p.szcode = s.szcode

GROUP BY p.userID

Join 可以建立不同數(shù)據(jù)流之間的連接,是大數(shù)據(jù)計算中的高級特性,復(fù)雜且代價大,但大多數(shù)場景都需要進行復(fù)雜的Join 操作.

Spark Streaming 會將逐條采集的數(shù)據(jù)按照事先設(shè)置好的批處理間隔匯總成一批數(shù)據(jù)進行處理,Join 操作是在每一個批數(shù)據(jù)上進行的,因此可通過對批處理間隔的合理設(shè)置避免Join 操作造成的運算復(fù)雜度較高.

2.3 評價指標(biāo)

針對流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試中評價指標(biāo)單一的問題,本文通過延遲、吞吐量、GC 時間等個方面對流式大數(shù)據(jù)處理系統(tǒng)進行評價.

延遲(Latency)是流式大數(shù)據(jù)處理中的一項常見且重要的指標(biāo),表示在處理過程中由于網(wǎng)絡(luò)或者計算產(chǎn)生的時間差.一般可以將延遲分為系統(tǒng)延遲和事件延遲[11].本文將延遲定義為數(shù)據(jù)從源段到輸出端所經(jīng)歷的非計算時間.

吞吐量(Throughput)是系統(tǒng)在單位時間內(nèi)的數(shù)據(jù)處理量.本文通過計算單位時間內(nèi)數(shù)據(jù)源端輸出的數(shù)據(jù)總量作為系統(tǒng)的吞吐量.

GC 時間(GC time)是系統(tǒng)執(zhí)行過程中垃圾回收機制的執(zhí)行時間.垃圾回收即遍歷應(yīng)用程序在Heap 上動態(tài)分配的所有對象,識別那些已經(jīng)死亡即不再被引用的對象,將該對象占用的內(nèi)存空間回收.垃圾回收的開銷是流式大數(shù)據(jù)處理系統(tǒng)內(nèi)存管理需要考慮的因素之一,本文通過GC 執(zhí)行時間衡量.

CPU 資源(CPU resources)即系統(tǒng)運行時的CPU 使用率.

Apache Spark Streaming 提供了Web UI 界面,在任務(wù)執(zhí)行過程中,可以實時查詢?nèi)蝿?wù)運行情況,便于測試指標(biāo)的查看和收集.本文借助Apache Spark Streaming提供的API 接口,實時收集運行時的延遲、吞吐量、GC 時間等評價指標(biāo)數(shù)據(jù),并以此進行實驗分析.

3 實驗驗證

本節(jié)對基于Apache Spark Streaming 實現(xiàn)的股票交易場景下的應(yīng)用進行了以下兩組測試:(1)測試數(shù)據(jù)輸入速率對系統(tǒng)性能的影響;(2)測試執(zhí)行器內(nèi)核數(shù)量對系統(tǒng)性能及擴展性的影響.通過對測試結(jié)果的分析,總結(jié)了系統(tǒng)的在不同的測試參數(shù)下的性能表現(xiàn).

3.1 實驗環(huán)境

為了模擬流式大數(shù)據(jù)處理系統(tǒng)在現(xiàn)實中的應(yīng)用,提高大數(shù)據(jù)處理的效率,實驗搭建了集群,部署了分布式計算環(huán)境用于Apache Spark Streaming 測試.測試集群由五臺機器組成,包括1 臺Master 節(jié)點和4 臺Slave 節(jié)點,共20 cores,集群架構(gòu)如圖2 所示,各個節(jié)點的配置信息如表2 所示.

圖2 集群架構(gòu)圖

表2 測試集群配置

3.2 實驗設(shè)計

實驗采用控制變量法,每次改變單一變量測試系統(tǒng)在不同情況下的性能表現(xiàn),每組進行五次測試,記錄平均值.本文針對Apache Spark Streaming 設(shè)計了兩個實驗,從數(shù)據(jù)輸入速率和任務(wù)并發(fā)度兩個方面考慮.

(1)通過控制socket 數(shù)據(jù)輸入端的線程休眠時間,測試系統(tǒng)輸入端的速率、執(zhí)行時間、任務(wù)數(shù)、延遲和GC 時間.

(2)通過控制每個執(zhí)行器的內(nèi)核數(shù),改變?nèi)蝿?wù)并發(fā)度,測試系統(tǒng)的擴展性,以及吞吐量、延遲和GC 時間的影響.

3.3 實驗結(jié)果及結(jié)論

(1)測試一:數(shù)據(jù)輸入速率對系統(tǒng)性能的影響

實驗通過Thread.sleep();函數(shù)控制socket 端向指定端口發(fā)送數(shù)據(jù)的速率,將線程的休眠時間分別設(shè)為500 毫秒,100 ms,50 ms,10 ms,1 ms 以及0,即每間隔500 ms,100 ms,50 ms,10 ms,1 ms 以及無間隔地發(fā)送數(shù)據(jù).同時,將數(shù)據(jù)的生成時間固定為5 min,實現(xiàn)對輸入端數(shù)據(jù)速率的控制.測試一的實驗結(jié)果如表3 所示.

表3 測試一實驗結(jié)果

通過實驗數(shù)據(jù)分析,可得結(jié)論如下:

1)發(fā)現(xiàn)一:當(dāng)數(shù)據(jù)輸入速率較高時,系統(tǒng)延遲呈現(xiàn)較大的增長.

實驗記錄了不同速率下的系統(tǒng)總延遲,如圖3 所示.可以發(fā)現(xiàn)隨著速率的增加,系統(tǒng)延遲總體呈上升趨勢的,這是符合邏輯的.但是在輸入速率較低、相差不大的情況下,系統(tǒng)延遲增加緩慢,控制在一定范圍之內(nèi),而當(dāng)輸入速率較高時,系統(tǒng)延遲會呈現(xiàn)一個較大的增長.

2)發(fā)現(xiàn)二:輸入速率的提高會使系統(tǒng)資源利用率增加.

隨著輸入速率的提高,實驗從GC 時間和CPU 資源兩個方面分析了系統(tǒng)資源利用率的變化.

圖3 不同速率下延遲比較

Apache Spark 默認(rèn)會將每個執(zhí)行器的60%的內(nèi)存空間用于緩存RDD,則在任務(wù)執(zhí)行期間,只有40%的內(nèi)存空間可以用來存放創(chuàng)建的對象.如果創(chuàng)建的對象過大,超過可用的內(nèi)存空間,就會觸發(fā)java JVM 的垃圾回收機制.從圖4 中可以看到,隨著輸入速率的提高,系統(tǒng)的GC 執(zhí)行時間也會增加,這是因為輸入速率的提高,任務(wù)數(shù)量及創(chuàng)建的對象也會增加,從而使GC 執(zhí)行次數(shù)增加,即系統(tǒng)運行時的內(nèi)存占用量增加.

圖4 不同速率下GC 時間比較

此外,隨著輸入速率的提高,CPU 利用率呈增長趨勢,CPU 負(fù)載逐步提升.但即使在速度達到最大時,master節(jié)點CPU 利用率平均為28.71%,最大可達48.47%,CPU 資源并未得到充分利用.

綜上所述,輸入速率的提高會使系統(tǒng)資源利用率增加,但在Socket 輸入最大值的情況下,仍未實現(xiàn)系統(tǒng)資源的充分利用.

3)發(fā)現(xiàn)三:數(shù)據(jù)輸入速率在一定閾值內(nèi),系統(tǒng)性能相對穩(wěn)定;數(shù)據(jù)輸入速率超過閾值時,系統(tǒng)性能下降.

通過對表3 實驗結(jié)果中的數(shù)據(jù)輸入速率與其他性能指標(biāo)的關(guān)系進行分析,可以發(fā)現(xiàn)當(dāng)數(shù)據(jù)輸入速率在19.30 records/s 到92.50 records/s 范圍內(nèi)時,程序執(zhí)行時間和GC 時間分別穩(wěn)定在8.5 s 和6 s,任務(wù)數(shù)和延遲波動較小.當(dāng)數(shù)據(jù)輸入速率提升到1701.06 records/s時,系統(tǒng)在執(zhí)行時間、任務(wù)數(shù)、延遲、GC 等性能指標(biāo)的度量上略有增長.然而,當(dāng)數(shù)據(jù)輸入速率上升到46 071.31 records/s 時,系統(tǒng)的執(zhí)行時間比數(shù)據(jù)輸入速率為1701.06 records/s 時增長了1 min,延遲增長了205 ms,執(zhí)行的任務(wù)數(shù)卻有所減少.由此可以得出結(jié)論:數(shù)據(jù)輸入速率在一定閾值內(nèi)時,系統(tǒng)的整體性能相對穩(wěn)定;當(dāng)數(shù)據(jù)輸入速率超過這一閾值時,系統(tǒng)性能會有所下降.

分析原因為隨著輸入速率的提高,系統(tǒng)接收的數(shù)據(jù)越來越多,系統(tǒng)的數(shù)據(jù)處理能力趨于飽和,甚至可能會出現(xiàn)計算過程中一個批次花費的時間大于系統(tǒng)設(shè)置的批處理間隔,這意味著數(shù)據(jù)接收速率大于數(shù)據(jù)處理速率,數(shù)據(jù)處理能力降低,系統(tǒng)性能也發(fā)生一定程度下降.但Spark Streaming 系統(tǒng)自身帶有反壓機制(Back Pressure),即使時間間隔內(nèi)無法完全處理當(dāng)前接收的數(shù)據(jù),也不會導(dǎo)致執(zhí)行器內(nèi)存泄漏.

此外,從圖5 中可以看出,任務(wù)數(shù)除了在速率從3.97 records/s 上升到19.3 records/s 時出現(xiàn)大幅度增長外,之后不再隨著數(shù)據(jù)輸入速率的增加發(fā)生較大變化,且在速率達到最大時出現(xiàn)下降,可見系統(tǒng)的任務(wù)數(shù)隨著數(shù)據(jù)輸入速率的增加出現(xiàn)瓶頸.

圖5 不同速率下任務(wù)數(shù)變化

(2)測試二:執(zhí)行器內(nèi)核數(shù)量對對系統(tǒng)性能及擴展性的影響

執(zhí)行器(executor)是Apache Spark 任務(wù)的執(zhí)行單元,運行在worker 上,是一組計算資源的集合.執(zhí)行器的內(nèi)核(core)數(shù)量可理解為執(zhí)行器的工作線程,實驗通過改變執(zhí)行器的內(nèi)核數(shù)控制系統(tǒng)的并發(fā)度.

實驗中共系統(tǒng)設(shè)置了4 個執(zhí)行器,將每個執(zhí)行器的內(nèi)核個數(shù)分別設(shè)置為2、4、8 和16,測試了2 min內(nèi)數(shù)據(jù)接收情況、系統(tǒng)總延遲以及GC 時間.測試二實驗結(jié)果如表4 所示.

表4 測試二實驗結(jié)果

通過實驗數(shù)據(jù)分析,可得結(jié)論如下:

1)發(fā)現(xiàn)一:執(zhí)行器的內(nèi)核數(shù)對系統(tǒng)吞吐量影響不大.

在數(shù)據(jù)輸入速率相同的情況下,隨著每個執(zhí)行器內(nèi)核個數(shù)的增加,系統(tǒng)在2 min 內(nèi)接收的數(shù)據(jù)整體呈現(xiàn)減少的趨勢,但從圖6 中可以看到,在考慮到網(wǎng)絡(luò)波動的情況下,系統(tǒng)接收數(shù)據(jù)的能力相似.因此,系統(tǒng)并發(fā)度的提升對Apache Spark Streaming 的吞吐量并沒有太大影響.

圖6 不同內(nèi)核數(shù)下的數(shù)據(jù)接收情況

2)發(fā)現(xiàn)二:執(zhí)行器內(nèi)核數(shù)量的增加可降低系統(tǒng)延遲.

結(jié)合結(jié)論一分析,在系統(tǒng)接收的數(shù)據(jù)量相差不大的情況下,隨著每個執(zhí)行器內(nèi)核個數(shù)的增加,系統(tǒng)的延遲會大幅度降低,內(nèi)核數(shù)為16 時的延遲只有內(nèi)核數(shù)為2 時的1/3,如圖7 所示.

分析原因為{任務(wù)執(zhí)行的并發(fā)度 = 執(zhí)行器的總數(shù)目 * 每個執(zhí)行器的內(nèi)核數(shù)},當(dāng)每個執(zhí)行器內(nèi)核數(shù)量增加時,任務(wù)并發(fā)度也會提高,多任務(wù)的并發(fā)執(zhí)行使得從而使系統(tǒng)延遲大幅度降低.可見系統(tǒng)并行度的提高使得Apache Spark Streaming 系統(tǒng)資源的利用率也隨之提高,系統(tǒng)在延遲上的擴展性良好.

圖7 不同內(nèi)核數(shù)下延遲比較

3)發(fā)現(xiàn)三:執(zhí)行器內(nèi)核數(shù)量的增加會造成系統(tǒng)資源利用率增加.

隨著內(nèi)核數(shù)量的增加,任務(wù)并發(fā)度提高,導(dǎo)致系統(tǒng)GC 時間增加,如圖8 所示.

圖8 不同內(nèi)核數(shù)下GC 時間比較

分析原因為內(nèi)核數(shù)量的增加提高了任務(wù)并發(fā)度,使得大量的對象會被創(chuàng)建,出發(fā)Java 垃圾回收機制的次數(shù)也會增加,從而使GC 時間增加.因此適當(dāng)減少內(nèi)核個數(shù)也是降低系統(tǒng)GC 開銷的一種方法.

此外,系統(tǒng)并發(fā)度的提高也使得CPU 利用率有所提高,在16 cores 時CPU 利用率平均為31.05%,最大可達35.35%.

綜上所述,執(zhí)行器內(nèi)核數(shù)的增加會提高系統(tǒng)并發(fā)度從而增加系統(tǒng)資源利用率.

4 結(jié)束語

本文設(shè)計并實現(xiàn)了一個流式大數(shù)據(jù)處理系統(tǒng)基準(zhǔn)測試框架,以Socket 作為流式數(shù)據(jù)生成,構(gòu)造股票高頻交易場景,并實現(xiàn)了GroupBy 和Join 兩個典型應(yīng)用,將實時計算與結(jié)構(gòu)化數(shù)據(jù)相結(jié)合.測試框架搭建在分布式集群環(huán)境下,選取了Apache Spark Streaming 作為待測系統(tǒng),從數(shù)據(jù)輸入速率和系統(tǒng)并行度兩個方面設(shè)計實驗,得到延遲、吞吐量、GC 時間等測試指標(biāo),以圖表的形式進行分析,發(fā)現(xiàn)流式大數(shù)據(jù)處理系統(tǒng)中出現(xiàn)的性能問題.實驗結(jié)果表明,隨著數(shù)據(jù)輸入速率的提高,系統(tǒng)性能保持相對穩(wěn)定,當(dāng)輸入速率達到一定閾值,系統(tǒng)會出現(xiàn)性能下降,資源利用率增加的現(xiàn)象;系統(tǒng)并行度的增加對吞吐量的影響較小,但系統(tǒng)延遲會大幅度降低,GC 時間有所增加,提高了系統(tǒng)資源的利用率.

未來將在以下幾個方面進行深入研究.一是可以將基準(zhǔn)測試框架應(yīng)用到不同的處理系統(tǒng)中,分析系統(tǒng)之間的差異,進行對比研究.二是對流式大數(shù)據(jù)處理系統(tǒng)的基準(zhǔn)測試不再僅限于對獨立的系統(tǒng),而是與第三方工具結(jié)合,如使用Kafka、Flume 等高級數(shù)據(jù)源產(chǎn)生數(shù)據(jù),模擬現(xiàn)實環(huán)境下的應(yīng)用.