決策樹模型預測Spark SQL作業執行時間的方法

吳 恩 慈

(上海淇毓信息科技有限公司 上海 200120)

0 引 言

Spark SQL借助核心引擎將集群規模擴展到數千個節點，Catalyst提供了基于規則和代價的優化器，把數據倉庫的計算能力推向新的高度。但在超大規模數據集上存在易用性和可擴展性的問題，SQL或Dataset程序在執行之前被解析成邏輯計劃，然后生成可執行的物理計劃，不同的執行計劃對性能有很大的影響。若能夠準確預測作業的執行時間，可以更加充分地運用作業運行時獲取的數據，動態地選擇一個最佳的物理執行計劃。例如基于作業運行時Shuffle信息，根據機器學習模型預測結果自動為每個作業設置合適的Shuffle Partition值，調整可能發生數據傾斜的任務并行度，動態地調整執行計劃，能夠有效提升集群性能。

1 相關工作

文獻[1]研究了預測Hadoop分布式存儲系統中查詢執行時間的方法，采用KCCA統計模型關聯查詢輸出大小與執行時間，通過最近的類似查詢的性能預測作業執行時間。文獻[2]提出了一種混合模型估計不同類型算子的性能，在關系型數據庫中有良好的表現。文獻[3]使用樣本數據模擬集群中不同節點的執行性能，存在Shuffle開銷的IO密集型作業的預測精度較低。文獻[4]提出了一種多元線性回歸方法，根據Spark性能指標預測集群性能，模型指標R2小于平均值。文獻[5]研究了更改Spark集群配置參數對性能的影響，在模型訓練階段捕獲關鍵性能指標，允許用戶詢問配置參數對性能的影響。文獻[6]通過分析MapReduce任務執行過程中收集的數據，提出兩階段回歸方法預測任務的完成時間。文獻[7]提出了一種Spark應用程序的預測框架，使用非負最小二乘法匹配給定數據的最佳模型和參數。

本文通過決策樹回歸模型預測作業執行時間的方法不同于上述文獻，本文基于Spark分布式計算引擎，提取作業特征與選擇指標時考慮了SQL作業存在的寬依賴、各階段間Shuffle開銷，以及集群的彈性配置。通過訓練決策樹及其組合算法生成的機器學習模型能夠準確地預測作業的執行時間，主要完成了三方面的工作：(1) 研究了作業調度策略與計算引擎，基于核心引擎構建機器學習應用模塊，使復雜的模型訓練與測試過程都變得易于實現。(2) 提供了模型的運行原理與特征數據的收集方法，通過Pipeline機制構建機器學習模型工作流。(3) 通過交叉驗證方法進行超參數調優，采用剪枝和組合算法優化過度擬合問題，確保在訓練集上訓練出泛化能力較強的模型。

2 調度策略與計算引擎

2.1 作業調度策略與流程

機器學習模型的訓練需要在多次迭代后獲得足夠小的誤差才會停止[8]。Spark任務調度策略簡化了機器學習模型訓練和測試的過程，DAG Scheduler實現了相關組件在內存中無縫集成與任務協調，根據依賴對Pipeline進行優化實現對作業并行計算。應用程序獲取計算資源的策略包括兩種，一種是把應用程序運行在盡可能多的Worker上，能夠充分使用集群資源，有利于數據處理的本地性；另一種是應用程序運行在盡可能少的Worker上，適合CPU密集型而內存使用較少的場景。如果任務處于作業開始的調度階段內，對應的RDD分區首選運行位置的數據本地性為Node Local；如果任務處于非作業開頭的調度階段，根據父調度階段運行的位置得到首選位置，若Executor處于活動狀態，數據本地性為Process Local。

圖1 作業調度流程

2.2 計算引擎與內存管理

Spark計算引擎在內存中完成多步驟的迭代計算，提升了機器學習應用的性能[9]。計算引擎主要包括執行內存與Shuffle兩部分，Shuffle性能優劣直接決定了計算引擎的性能和吞吐量。執行內存包括在JVM堆上分配的執行內存池和在操作系統內存中分配的Tungsten。操作系統內存是整個架構的基礎，無論執行內存如何分配，都離不開系統內存的支持。操作系統通過確定數據所在的頁緩存，并使用頁緩存的偏移量和數據的長度讀取數據，避免了JVM加載額外類文件和對象，降低了GC掃描與回收的頻率。內存管理器負責申請和釋放執行內存，執行內存的消費者組件包括3種，任務中間過程輸出數據在JVM堆上進行緩存和聚合等處理，操作系統內存中進行溢出和持久化處理的Shuffle操作，以及鍵值對存儲到連續的內存塊的批處理。

圖2為內存申請與分配原理圖。Tungsten是一種內存分配與釋放的實現方式，直接操作系統內存實現類似于操作系統的頁緩存的Memory Block數據結構，準確地申請和釋放堆外內存、計算序列化數據占用的空間，降低了管理的難度和誤差[10]。內存塊中的數據位于JVM堆內存或者堆外內存，主要包括obj、offset和length三個屬性。obj屬性保存了對象在JVM堆中的地址，offset屬性保存了頁緩存的起始地址相對于對象在JVM堆中地址的偏移量，length屬性保存了頁緩存的大小。Tungsten處于堆內內存模式時，數據作為對象存儲在JVM堆中，從堆內找到對象使用offset定位數據的具體位置。處于堆外內存模式時，通過offset屬性從堆外內存中定位數據，從obj和offset定位的起始位置開始獲取固定長度的連續內存塊。若申請的內存塊大于等于1 MB，且Memory Buffer Pools中存在指定大小的內存塊時，從內存緩存池中獲取，否則單獨創建內存塊用于分配。

圖2 內存申請與分配原理

3 預測模型構建

3.1 預測模型的運行原理

圖3為作業執行時間預測原理圖。預測模型的響應時間需要滿足在線預測的實時性要求，能夠并行化操作和處理更大規模的數據集與更加復雜的邏輯。功能包括收集作業運行時特征與指標、構建模型訓練與測試過程、預測作業的執行時間。作業被編譯生成DAG時，在調度程序中注入計數器計算每個階段所需特征。運行SQL作業實現模型離線訓練，訓練階段的目標是從作業中提取特征向量X，以及相應的性能度量指標向量Y。如果一個階段有N個Task并行執行將為該階段收集N組特征，特征收集器的輸出是特征向量X。特征類別包括數據集規模、Shuffle和IO特征，以及集群配置等。模型的輸入包括特征向量和度量指標，輸出是特征和指標之間關系的回歸模型，通過比較實際值和預測值評估模型預測的準確性。

圖3 作業執行時間預測原理

通過解析作業執行計劃提取特征向量，機器學習模型預測執行計劃中每個階段執行時間，根據每個階段的執行時間估計作業的執行時間。DAG中階段是根據RDD依賴關系運行，某個階段的執行時間是此階段開始的最早任務的開始時間，與最后執行任務的完成時間之間的間隔，如式(1)所示。FTST表示在數據分區上某個階段第一個任務的開始執行時間，LTET表示在某個階段最后一個任務的結束時間。式(2)給出如何根據每個階段的估計執行時間預測整個作業的執行時間，其中作業有N個階段。

TStage(i)=FTSFi-LTETi

(1)

(2)

3.2 特征與指標的收集方法

表1為Task特征向量X來源與取值示例。機器學習任務中特征選擇是重要的數據預理過程，剔除冗余和無關的特征，在高維數據分析中可以提升機器學習的性能。特征選擇方法主要分為監督和無監督兩種。卡方選擇是統計學上常用的特征選擇方法，通過特征和真實標簽之間執行卡方檢驗確定關聯度。編號1～3表征了網絡流量特征，編號4～11表征了作業執行過程中的Shuffle和IO特征，編號12～15表征了數據規模特征。作業特征與指標數據的提取主要借助Listener Bus機制和Metrics System，采用異步線程將事件提交到對應的事件監聽器。作業數據集獲取方法包括三種：運行時作業調度頁面、通過REST接口獲取度量信息、借助外部監控工具。本文通過REST接口方式提取相關數據。

表1 特征向量來源與取值

表2為Task度量指標Y來源與取值示例。選取的度量指標能夠體現不同規模的數據集、不同類型算子的Shuffle和IO操作，以及網絡流量方面的時間開銷。充分考慮了作業執行時間指標的動態性，每個作業執行計劃的可用資源不同，作業并行運行時存在資源競爭，作業的GC時間和數據序列化與反序列化時間，以及網絡傳輸存在一定的隨機性與關聯性，例如存在Shuffle過程算子的IO操作往往比較耗時。

表2 度量指標來源與取值

3.3 預測模型的評估指標

決策樹可以使用復雜的非線性模型擬合數據，通過改變不純度的度量方法用于回歸分析。類似線性回歸模型使用對應的損失函數，決策樹用于回歸時使用不純度度量方法[13]。方差是用于量度回歸模型的節點處標簽均勻性的量度。如式(3)所示，均方根誤差RMSE是均方誤差MSE的平方根，精確度會進一步放大，越接近零表示預測越準確，wTx(i)是預測值，y(i)是所有實際值。如式(4)所示，平均絕對誤差MAE是預測值與實際值之差的絕對值的平均值，MAE避免正負誤差相互抵消，更好地反映預測值誤差的實際情況。如式(5)所示，擬合優度(R2)用來評估模型擬合數據的好壞程度，測量目標變量的變異度，表示可以根據自變量的變化來解釋因變量的變體部分[14]。R2越接近1，表示自變量對因變量的解釋程度越高。

(3)

(4)

(5)

4 過度擬合問題

4.1 過度擬合原因與對策分析

決策樹易于解釋和無須特征縮放，是分類和回歸機器學習的有效方法[11]。決策樹生成是一種貪婪算法，通過從一組可能的分割中選擇最佳分割，貪婪地選擇每個分區，以最大化樹節點處的信息增益[12]。從根節點開始計算節點所有可能特征的信息增益，選擇信息增益最大的特征，然后遞歸調用上述方法構造子節點直到所有特征的信息增益不再增加。當決策樹中節點下的所有記錄屬于同一個類，或者所有記錄屬性具有相同值時，生長過程終止。遞歸算法生成的決策樹通常具有過擬合問題，該問題可以從兩個方面進行優化。通過剪枝主動去掉一些分支來降低過擬合的風險，該方法的關鍵是如何確定最優參數，通過調整參數提前在節點處停止遞歸構造能夠有效避免過度擬合，可調整的參數包括Max Depth與Min Info Gain等。另一種方法是通過組合算法在一定程度上抵消過度擬合，Spark ML支持Random Forest和Gradient Boosted Trees兩種決策樹組合算法。

云南省統計局相關負責人分析認為， 投資是拉動經濟增長的關鍵因素。今年以來云南省民間投資、制造業投資、服務業投資增長明顯，促進了投資的高質量發展。下一步，應著力促進投資增長集中在高技術行業、高端裝備制造業以及消費轉型升級重要行業。同時，要進一步推進民間投資發展，降低準入門檻，優化營商環境。

4.2 交叉驗證法優化模型參數

通過最小化決策樹的損失函數實現剪枝，在一定程度上能夠避免過度擬合，當節點深度等于Max Depth參數時決策樹停止生長。Min Info Gain參數是拆分必須改善信息增益的最小值，沒有分割候選項導致信息增益大于最小值時決策樹停止生長。機器學習能夠通過數據集找到特定問題的最佳超參數，可以在獨立的Estimator中完成，或者在包含多種算法和特征選擇的工作流中完成。通過重復進行模型訓練和測試步驟，采用隨機化抽樣方法選定K個相似的互斥子集，每個子集盡可能地保持數據分布的一致性，分別訓練和測試模型，通過取K個模型的均值避免過度擬合問題，該過程稱為交叉驗證，評估結果的穩定性和保真性在很大程度上依賴K的取值。

采用交叉驗證方法評估模型參數時，計算由估計器擬合模型的不同數據對的平均評估指數，使用此參數重新擬合整個數據集的估算器，找到最佳配置參數，在整個訓練集上訓練具有較強泛化能力和相對較小誤差的最佳模型。當交叉驗證方法的訓練成本較高時，可使用Train Validation Split方法進行超參數調整，創建單個訓練和測試數據集對。使用訓練比例參數將數據集拆分為兩部分，常用75%生成一組訓練和測試數據集對，使用最佳參數配置和完整數據集擬合估計器。交叉驗證方法每個參數執行K次評估，Train Validation Split方法每個參數組合僅評估一次，訓練數據集不夠大時結果的可靠性較低。

4.3 組合算法優化過度擬合

隨機森林采用Bagging思想，并行訓練一組決策樹集，模型訓練過程是隨機的，在每次迭代時對原始數據集進行二次采樣，在每個樹節點處分割不同的隨機特征子集。先對節點隨機選擇包含K個屬性的子集再選擇最優屬性，參數控制了隨機性的引入程度。隨機森林并不基于模型殘差來構建集成模型，能夠取得較低的方差，預測結果聚合了決策樹集的預測。將每棵樹的預測計為一類投票，獲得投票最多的類別作為分類預測結果，平均值用作預測回歸結果，能夠捕獲非線性特征提高判別精度，在一定程度上避免過擬合。

梯度提升樹算法是迭代訓練決策樹，比隨機森林需要更長的訓練時間。每次迭代使用當前數據集來預測每個訓練實例的標簽，將預測結果與真實標簽進行比較，然后重新標記數據集[15]。在下一次迭代訓練中決策樹將糾正先前的偏差，并可以進一步減少每次迭代訓練數據的偏差，重新標記實例的機制由損失函數確定。表3為Spark ML梯度提升樹支持的損失函數類型，N表示實例數，yi是實例的標簽，xi是實例的特征，F(xi)是模型預測標簽。梯度提升樹根據觀測值對預測結果進行調整，容易受到噪聲點的影響，在訓練時使用Run With Validation方法進行驗證，當驗證錯誤的改進不超過策略設置的容差時停止訓練，能夠有效地防止過度擬合。

表3 梯度提升樹損失函數

5 實 驗

5.1 SQL作業實驗過程

實驗環境的集群配置11個計算節點，每個節點配置16內核、32 GB內存、512 GB SSD、萬兆網卡。每個節點上以獨立模式運行Spark 2.0，數據存儲在HDFS 2.6。決策樹回歸模型的超參數配置如表4所示，包括模型訓練、度量方法和緩存策略等，超參數是通過交叉驗證方法提取的最優參數。調整Max Depth與Min Info Gain訓練參數提前在節點處停止遞歸構造避免過度擬合。

表4 決策樹回歸模型超參數

圖4為使用Spark Bench模擬Spark SQL作業生成測試數據集規模信息，數據集包含約1 000萬個實例，每個實例有50個屬性。測試設定的應用場景與實際業務場景類似，SQL查詢使用模擬生成的電子商務系統的訂單數據，涵蓋了CPU、內存、Shuffle和IO密集型工作負載。在此數據集上進行Spark SQL工作負載測試，包括Join與Group by算子的實驗。

圖4 Spark Bench模擬數據集

圖5為Join算子作業數據集，數據集中用于模型訓練的比率是0.7。Join算子作業共產生個236個Task，涉及到Shuffle Read(Write)數據量約1 800 MB。計算過程中Input(Output)數據規模約1 900 MB，作業累計持續時間5 min。

圖5 Join算子作業數據集

圖6為Group by算子作業數據集，數據集中用于模型訓練的比率是0.7。Group by算子作業共產生218個Task，涉及到Shuffle Read(Write)數據量約1 800 MB。計算過程中Input(Output)數據規模約1 600 MB，作業累計持續時間27 s。

圖6 Group by算子作業數據集

5.2 預測模型準確率分析

表5展示了決策樹及其組合算法回歸模型評估指標，其中Join算子的3種決策樹回歸模型的預測精度有明顯的區別，梯度提升樹回歸模型的R2超過0.8，說明訓練階段選定的特征和模型有較好的匹配。隨機森林的指標存在一定的偏差，但優于決策樹。三種回歸模型RMSE與MAE指標偏差也呈現相同的規律。Join算子模型預測值與實際值對比如圖7所示，其中：點號(·)表示真實值。加號(+)表示預測值，可以觀察到梯度提升樹回歸模型的擬合程度優于隨機森林模型與決策樹模型。

表5 決策樹回歸模型評估指標

由表5所示的Group by算子的三種決策樹回歸模型指標可見，隨機森林和梯度提升樹的R2都超過0.8，決策樹的擬合程度較低，RMSE與MAE指標偏差也呈現出相同的規律。Group by算子模型預測值與實際值對比如圖8所示，梯度提升樹和隨機森林的擬合程度相近，所選擇的特征很好地表征運算符，決策樹有較大的偏差。

(a) 決策樹

三種其他回歸模型評估指標如表6所示，線性回歸使用彈性網正則化模型，廣義線性回歸模型采用高斯響應和身份鏈接功能訓練模型。Join算子三種回歸模型的R2都小于0.6，Group by算子三種回歸模型的R2都小于0.3，其中廣義線性回歸模型的指標要優于其他兩種，但都低于決策樹回歸模型的指標。Join算子線性回歸模型預測值與實際值如圖9所示；Group by算子線性回歸模型預測值與實際值如圖10所示。可以看出，所選的特征數據與模型的匹配度較差。

表6 線性回歸模型評估指標

(a) 邏輯回歸

(a) 邏輯回歸

6 結 語

通過收集表征Spark SQL作業特征的數據集和時間指標，訓練決策樹及其組合方法的回歸模型，評估實驗表明梯度提升樹回歸模型在預測Spark SQL作業執行時間上有較好的表現，評估指標明顯優于線性回歸模型。通過Join算子與Group by算子的數據集測試，梯度提升樹回歸模型的R2都超過0.8，指標RMSE與MAE也達到預期效果，且滿足在線預測的實時性要求，表明模型預測Spark SQL作業的執行時間的方法具有準確性和實用性。下一步將在此基礎上，根據預測模型的輸出結果，充分運用作業運行時數據，自適應地調整相關參數，選擇最優的作業執行計劃，提交給DAG Scheduler執行，提升集群性能。