劉思宇+梁毅+陳誠

摘要：Spark是一種新型分布式海量數(shù)據(jù)處理平臺，在應(yīng)用執(zhí)行過程中，Spark以任務(wù)作為最小執(zhí)行單元。因此，任務(wù)執(zhí)行時(shí)間預(yù)測是指導(dǎo)Spark進(jìn)行性能分析、優(yōu)化資源調(diào)度以及故障監(jiān)控的基礎(chǔ)。在Spark平臺中，由于計(jì)算數(shù)據(jù)分布不均及網(wǎng)絡(luò)資源的共享，導(dǎo)致同樣計(jì)算邏輯的任務(wù)在不同計(jì)算節(jié)點(diǎn)上執(zhí)行的時(shí)間可能產(chǎn)生很大差異，需根據(jù)實(shí)時(shí)運(yùn)行環(huán)境進(jìn)行動態(tài)預(yù)測。通過結(jié)合任務(wù)在不同節(jié)點(diǎn)所需數(shù)據(jù)量以及集群網(wǎng)絡(luò)狀況，對任務(wù)在不同節(jié)點(diǎn)的執(zhí)行時(shí)間進(jìn)行預(yù)測。實(shí)驗(yàn)表明，該方法對任務(wù)進(jìn)行預(yù)估，誤差可保證在19%以內(nèi)，任務(wù)執(zhí)行時(shí)間預(yù)估算法對Spark調(diào)優(yōu)有一定的指導(dǎo)作用。

關(guān)鍵詞：大數(shù)據(jù)；Spark；預(yù)測；分布式；任務(wù)

DOIDOI：10.11907/rjdk.171509

中圖分類號：TP306

文獻(xiàn)標(biāo)識碼：A 文章編號：1672-7800（2017）012-0019-03

Abstract：Spark is a new distributed big data processing platform. In the implementation of Spark， task is the minimum execution unit. Therefore， the prediction of the execution time of the task can guide Spark to perform performance analysis， optimize resource scheduling and fault monitoring. In Spark platform， due to the uneven distribution of computing data and the sharing of network resources， the task with same computing logic may have different execution time in different nodes， and it needs to be dynamically predicted according to the real time environment. Currently on the Spark platform， the prediction technology is rarely studied. This paper predicts the execution time of each task at different nodes by combing the amount of data required by different tasks in each node and the status of cluster network. Experiment show that the method can be used to estimate the tasks in the task set， the error can guarantee less than 19%. Therefore， the task execution time estimation algorithm proposed in this paper has some guiding effect on Spark tuning.

Key Words：big data； Spark； prediction； distributed； task

0 引言

大數(shù)據(jù)時(shí)代，新型海量數(shù)據(jù)處理平臺大量涌現(xiàn)。其中，以Spark數(shù)據(jù)處理平臺為典型代表的分布式內(nèi)存計(jì)算平臺得到廣泛關(guān)注 [1]。Spark是繼Hadoop之后提出的一種基于內(nèi)存的分布式大數(shù)據(jù)處理平臺，被譽(yù)為可以取代Map/Reduce的下一代大數(shù)據(jù)處理核心技術(shù)[2-3]。與Hadoop Map/Reduce相比，Spark基于內(nèi)存的運(yùn)算可提升100倍處理速度[4]。

任務(wù)是Spark的最小執(zhí)行單元。由于不同任務(wù)所需數(shù)據(jù)可能存在于集群各節(jié)點(diǎn)上，且數(shù)據(jù)量不盡相同，導(dǎo)致同樣計(jì)算邏輯任務(wù)在不同的計(jì)算節(jié)點(diǎn)上執(zhí)行的時(shí)間產(chǎn)生很大差異，需要根據(jù)應(yīng)用運(yùn)行的實(shí)時(shí)環(huán)境進(jìn)行動態(tài)預(yù)測[5]。對任務(wù)執(zhí)行時(shí)間的有效預(yù)測可以指導(dǎo)Spark進(jìn)行性能分析、優(yōu)化資源調(diào)度以及監(jiān)控平臺故障。目前，在Spark平臺還沒有任務(wù)執(zhí)行時(shí)間預(yù)估技術(shù)。

本文分析了Spark平臺任務(wù)的拉取、執(zhí)行過程，結(jié)合各任務(wù)在不同節(jié)點(diǎn)所需的數(shù)據(jù)量以及集群網(wǎng)絡(luò)狀況，對各任務(wù)執(zhí)行時(shí)間進(jìn)行預(yù)測，為優(yōu)化資源調(diào)度、應(yīng)用性能分析奠定基礎(chǔ)。

1 相關(guān)技術(shù)

1.1 Spark系統(tǒng)模型

Spark 是由 UC Berkeley AMP 實(shí)驗(yàn)室開發(fā)的開源通用的海量數(shù)據(jù)處理平臺，是對Map/Reduce型海量數(shù)據(jù)處理平臺的創(chuàng)新與豐富，其構(gòu)架如圖1所示。

Spark平臺采用Master/Slave結(jié)構(gòu)，其中集群管理器作為Master端，負(fù)責(zé)平臺中的應(yīng)用與資源管理；計(jì)算節(jié)點(diǎn)作為Slave端，負(fù)責(zé)啟動任務(wù)執(zhí)行器Executor，由Executor負(fù)責(zé)任務(wù)的實(shí)際計(jì)算。在執(zhí)行應(yīng)用時(shí)，會在相應(yīng)節(jié)點(diǎn)運(yùn)行Driver進(jìn)程，負(fù)責(zé)整個(gè)應(yīng)用的執(zhí)行和管理。

Spark大數(shù)據(jù)處理平臺的計(jì)算模型使用有向無環(huán)圖（Directed Acyclic Graph， DAG），描述復(fù)雜數(shù)據(jù)處理邏輯并提供更豐富的數(shù)據(jù)操作原語。Spark引入新的分布式數(shù)據(jù)集抽象表達(dá)模型——彈性分布式數(shù)據(jù)集（Resilient Distributed Datasets，RDDs）。RDD作為Spark平臺的核心概念，用于描述分布存儲于多個(gè)節(jié)點(diǎn)的海量數(shù)據(jù)集。

1.2 Spark任務(wù)執(zhí)行模型

Spark計(jì)算模型中，依據(jù)數(shù)據(jù)操作類型，作業(yè)分為多個(gè)階段（Stage），各階段以一定的拓?fù)浣Y(jié)構(gòu)執(zhí)行。在單個(gè)階段內(nèi)部，為了操作海量數(shù)據(jù)集，Spark會并行執(zhí)行一組完全相同的任務(wù)來處理RDD的每一分片。任務(wù)執(zhí)行完成后，輸出結(jié)果會進(jìn)行分區(qū)處理，具有相同分區(qū)值的元組會傳給后繼階段中同一個(gè)任務(wù)進(jìn)行相關(guān)計(jì)算。不同Stage之間的數(shù)據(jù)傳輸過程被稱為混洗（Shuffle）。同一階段任務(wù)啟動時(shí)，會首先選擇合適的節(jié)點(diǎn)，并拉取其它節(jié)點(diǎn)上屬于該任務(wù)的數(shù)據(jù)進(jìn)行計(jì)算。

執(zhí)行任務(wù)時(shí)，Spark首先并行拉取遠(yuǎn)程數(shù)據(jù)，而后對拉取來的數(shù)據(jù)進(jìn)行處理，Spark的任務(wù)執(zhí)行模型如圖2所示。

2 任務(wù)執(zhí)行時(shí)間預(yù)測方法

本節(jié)介紹任務(wù)執(zhí)行時(shí)間預(yù)測技術(shù)，主要是不同節(jié)點(diǎn)上各任務(wù)數(shù)據(jù)量獲取及各任務(wù)執(zhí)行時(shí)間計(jì)算。

2.1 各節(jié)點(diǎn)數(shù)據(jù)量獲取

當(dāng)新的Stage開始時(shí)，各任務(wù)需要根據(jù)上一階段不同任務(wù)寫入Spark既有模塊MapStatus中的文件位置信息，尋找合適的啟動位置，本方案擬定義新的方法對Map任務(wù)輸出到各節(jié)點(diǎn)的數(shù)據(jù)量進(jìn)行統(tǒng)計(jì)，核心算法如下：

2.3 系統(tǒng)實(shí)現(xiàn)

系統(tǒng)基于Spark 1.6平臺實(shí)現(xiàn)，主要針對Spark平臺中既有的各模塊進(jìn)行改造，并新增了數(shù)據(jù)采集模塊和任務(wù)執(zhí)行代價(jià)預(yù)測模塊，圖3為系統(tǒng)架構(gòu)。

在Executor端，新增了用以獲取Executor間網(wǎng)絡(luò)通信狀況以及Executor輸出數(shù)據(jù)信息的ExecutorMonitor。

在Driver端，新增了PartitionSizeGetter組件以及TaskCostEstimator組件。其中PartitionSizeGetter模塊負(fù)責(zé)在任務(wù)啟動之前訪問Spark既有的組件MapOutputTracker，統(tǒng)計(jì)出該任務(wù)在不同節(jié)點(diǎn)所需拉取的數(shù)據(jù)量，為任務(wù)執(zhí)行代價(jià)估計(jì)模塊做準(zhǔn)備。TaskCostEstimator通過PartitionSizeGetter模塊獲取不同任務(wù)在不同節(jié)點(diǎn)上所需的數(shù)據(jù)量，以及在Driver上的網(wǎng)絡(luò)通信狀況，通過計(jì)算模型進(jìn)行綜合，對任務(wù)在不同位置的執(zhí)行時(shí)間進(jìn)行預(yù)估。

3 性能評估

3.1 實(shí)驗(yàn)環(huán)境及負(fù)載選擇

本系統(tǒng)基于Apache Spark 1.6實(shí)現(xiàn)，所用操作系統(tǒng)為CentOS6.2，實(shí)驗(yàn)測試環(huán)境由 4臺物理節(jié)點(diǎn)構(gòu)成，每臺節(jié)點(diǎn)的硬件配置、網(wǎng)絡(luò)狀況、操作系統(tǒng)以及JVM版本等如表1所示。在本測試環(huán)境中，1臺節(jié)點(diǎn)作為提交節(jié)點(diǎn)，其余3臺作為數(shù)據(jù)計(jì)算節(jié)點(diǎn)。

任務(wù)執(zhí)行時(shí)間通過Spark提供的UI界面監(jiān)控獲得，選擇BigDataBench中的標(biāo)準(zhǔn)負(fù)載WordCount進(jìn)行實(shí)驗(yàn)。

3.2 預(yù)測結(jié)果分析

監(jiān)測WordCount負(fù)載中執(zhí)行Count操作的各任務(wù)所花費(fèi)的時(shí)間，圖4為任務(wù)執(zhí)行時(shí)間預(yù)測值與實(shí)際值的對比情況。

使用本文方法對各任務(wù)的執(zhí)行時(shí)間進(jìn)行預(yù)測，所得到的預(yù)測曲線與實(shí)際曲線基本吻合，任務(wù)預(yù)測時(shí)間與實(shí)際執(zhí)行時(shí)間的平均誤差為13%，最大誤差不超過19%。

4 結(jié)語

本文面向Spark海量數(shù)據(jù)處理平臺，設(shè)計(jì)并實(shí)現(xiàn)了任務(wù)執(zhí)行時(shí)間的預(yù)測模型，該模型充分考慮了任務(wù)的數(shù)據(jù)拉取代價(jià)與數(shù)據(jù)處理代價(jià)，對任務(wù)執(zhí)行時(shí)間進(jìn)行了綜合預(yù)測。實(shí)驗(yàn)表明，本文預(yù)測方法可有效預(yù)測任務(wù)執(zhí)行時(shí)間，最大誤差不超過19%。

參考文獻(xiàn)：

[1] AVAILABLE.Apache spark [EB/OL]. https：//spark.apache.org.

[2] AVAILABLE.Apache hadoop [EB/OL]. http：//hadoop.apache.org/.

[3] DEAN J， GHEMAWAT S. MapReduce： simplified data processing on large clusters[EB/OL]. http：//blog.csdn.net/cnlht/article/details/6181799.

[4] KRISH K R， ANWAR A， BUTT A R. HatS： a heterogeneity-aware tiered storage for hadoop[C].Ieee/acm International Symposium on Cluster， Cloud and Grid Computing. IEEE，2014：502-511.

[5] ZAHARIA M， CHOWDHURY M， DAS T， et al. Resilient distributed datasets[C].A fault-tolerant abstraction for in-memory cluster computing，2014.

（責(zé)任編輯：杜能鋼）