4. 結(jié)合圖解Spark書進(jìn)行應(yīng)用與優(yōu)化

《圖解Spark：核心技術(shù)與案例實(shí)戰(zhàn)》一書以Spark2.0版本為基礎(chǔ)進(jìn)行編寫，系統(tǒng)介紹了Spark核心及其生態(tài)圈組件技術(shù)。其內(nèi)容包括Spark生態(tài)圈、實(shí)戰(zhàn)環(huán)境搭建和編程模型等，重點(diǎn)介紹了作業(yè)調(diào)度、容錯(cuò)執(zhí)行、監(jiān)控管理、存儲(chǔ)管理以及運(yùn)行架構(gòu)，同時(shí)還介紹了Spark生態(tài)圈相關(guān)組件，包括了Spark SQL的即席查詢、Spark Streaming的實(shí)時(shí)流處理、MLlib的機(jī)器學(xué)習(xí)、GraphX的圖處理和Alluxio的分布式內(nèi)存文件系統(tǒng)等。下面介紹京東預(yù)測系統(tǒng)如何進(jìn)行資源調(diào)度，并描述如何使用Spark存儲(chǔ)相關(guān)知識(shí)進(jìn)行系統(tǒng)優(yōu)化。

4.1 結(jié)合系統(tǒng)中的應(yīng)用

在圖解Spark書的第六章描述了Spark運(yùn)行架構(gòu)，介紹了Spark集群資源調(diào)度一般分為粗粒度調(diào)度和細(xì)粒度調(diào)度兩種模式。粗粒度包括了獨(dú)立運(yùn)行模式和Mesos粗粒度運(yùn)行模式，在這種情況下以整個(gè)機(jī)器作為分配單元執(zhí)行作業(yè)，該模式優(yōu)點(diǎn)是由于資源長期持有減少了資源調(diào)度的時(shí)間開銷，缺點(diǎn)是該模式中無法感知資源使用的變化，易造成系統(tǒng)資源的閑置，從而造成了資源浪費(fèi)。而細(xì)粒度包括了Yarn運(yùn)行模式和Mesos細(xì)粒度運(yùn)行模式,該模式的優(yōu)點(diǎn)是系統(tǒng)資源能夠得到充分利用，缺點(diǎn)是該模式中每個(gè)任務(wù)都需要從管理器獲取資源，調(diào)度延遲較大、開銷較大。

由于京東Spark集群屬于基礎(chǔ)平臺(tái)，在公司內(nèi)部共享這些資源，所以集群采用的是Yarn運(yùn)行模式，在這種模式下可以根據(jù)不同系統(tǒng)所需要的資源進(jìn)行靈活的管理。在YARN-Cluster模式中，當(dāng)用戶向YARN集群中提交一個(gè)應(yīng)用程序后，YARN集群將分兩個(gè)階段運(yùn)行該應(yīng)用程序：第一個(gè)階段是把Spark的SparkContext作為Application Master在YARN集群中先啟動(dòng)；第二個(gè)階段是由Application Master創(chuàng)建應(yīng)用程序，然后為它向Resource Manager申請資源，并啟動(dòng)Executor來運(yùn)行任務(wù)集，同時(shí)監(jiān)控它的整個(gè)運(yùn)行過程，直到運(yùn)行完成。下圖為Yarn-Cluster運(yùn)行模式執(zhí)行過程：

 4.2 結(jié)合系統(tǒng)的優(yōu)化

我們都知道大數(shù)據(jù)處理的瓶頸在IO。我們借助Spark可以把迭代過程中的數(shù)據(jù)放在內(nèi)存中，相比MapReduce寫到磁盤速度提高近兩個(gè)數(shù)量級；另外對于數(shù)據(jù)處理過程盡可能避免Shuffle，如果不能避免則Shuffle前盡可能過濾數(shù)據(jù)，減少Shuffle數(shù)據(jù)量；最后，就是使用高效的序列化和壓縮算法。在京東預(yù)測系統(tǒng)主要就是圍繞這些環(huán)節(jié)展開優(yōu)化，相關(guān)Spark存儲(chǔ)原理知識(shí)可以參見圖解Spark書第五章的詳細(xì)描述。

由于資源限制，分配給預(yù)測系統(tǒng)的Spark集群規(guī)模并不是很大,在有限的資源下運(yùn)行Spark應(yīng)用程序確實(shí)是一個(gè)考驗(yàn)，因?yàn)樵谶@種情況下經(jīng)常會(huì)出現(xiàn)諸如程序計(jì)算時(shí)間太長、找不到Executor等錯(cuò)誤。我們通過調(diào)整參數(shù)、修改設(shè)計(jì)和修改程序邏輯三個(gè)方面進(jìn)行優(yōu)化：

4.2.1 參數(shù)調(diào)整

• 減少num-executors，調(diào)大executor-memory，這樣的目的是希望Executor有足夠的內(nèi)存可以使用。
• 查看日志發(fā)現(xiàn)沒有足夠的空間存儲(chǔ)廣播變量，分析是由于Cache到內(nèi)存里的數(shù)據(jù)太多耗盡了內(nèi)存，于是我們將Cache的級別適當(dāng)調(diào)成MEMORY_ONLY_SER和DISK_ONLY。
• 針對某些任務(wù)關(guān)閉了推測機(jī)制，因?yàn)橛行┤蝿?wù)會(huì)出現(xiàn)暫時(shí)無法解決的數(shù)據(jù)傾斜問題，并非節(jié)點(diǎn)出現(xiàn)問題。
• 調(diào)整內(nèi)存分配，對于一個(gè)Shuffle很多的任務(wù)，我們就把Cache的內(nèi)存分配比例調(diào)低，同時(shí)調(diào)高Shuffle的內(nèi)存比例。

 4.2.2 修改設(shè)計(jì)

參數(shù)的調(diào)整雖然容易做，但往往效果不好，這時(shí)候需要考慮從設(shè)計(jì)的角度去優(yōu)化：

• 原先在訓(xùn)練數(shù)據(jù)之前會(huì)先讀取歷史的幾個(gè)月甚至幾年的數(shù)據(jù)，對這些數(shù)據(jù)進(jìn)行合并、轉(zhuǎn)換等一系列復(fù)雜的處理，最終生成特征數(shù)據(jù)。由于數(shù)據(jù)量龐大，任務(wù)有時(shí)會(huì)報(bào)錯(cuò)。經(jīng)過調(diào)整后當(dāng)天只處理當(dāng)天數(shù)據(jù)，并將結(jié)果保存到當(dāng)日分區(qū)下，訓(xùn)練時(shí)按天數(shù)需要讀取多個(gè)分區(qū)的數(shù)據(jù)做union操作即可。
• 將“模型訓(xùn)練”從每天執(zhí)行調(diào)整到每周執(zhí)行，將“模型參數(shù)選取”從每周執(zhí)行調(diào)整到每月執(zhí)行。因?yàn)檫@兩個(gè)任務(wù)都十分消耗資源，并且屬于不需要頻繁運(yùn)行，這么做雖然準(zhǔn)確度會(huì)略微降低，但都在可接受范圍內(nèi)。
• 通過拆分任務(wù)也可以很好的解決資源不夠用的問題。可以橫向拆分，比如原先是將100個(gè)品類數(shù)據(jù)放在一個(gè)任務(wù)中進(jìn)行訓(xùn)練，調(diào)整后改成每10個(gè)品類提交一次Spark作業(yè)進(jìn)行訓(xùn)練。這樣雖然整體執(zhí)行時(shí)間變長，但是避免了程序異常退出，保證任務(wù)可以執(zhí)行成功。除了橫向還可以縱向拆分，即將一個(gè)包含10個(gè)Stage的Spark任務(wù)拆分成兩個(gè)任務(wù)，每個(gè)任務(wù)包含5個(gè)Stage，中間數(shù)據(jù)保存到HDFS中。

4.2.3 修改程序邏輯

為了進(jìn)一步提高程序的運(yùn)行效率，通過修改程序的邏輯來提高性能，主要是在如下方面進(jìn)行了改進(jìn)：避免過多的Shuffle、減少Shuffle時(shí)需要傳輸?shù)臄?shù)據(jù)和處理數(shù)據(jù)傾斜問題等。

1. 避免過多的Shuffle

Spark提供了豐富的轉(zhuǎn)換操作，可以使我們完成各類復(fù)雜的數(shù)據(jù)處理工作，但是也正因?yàn)槿绱宋覀冊趯慡park程序的時(shí)候可能會(huì)遇到一個(gè)陷阱，那就是為了使代碼變的簡潔過分依賴RDD的轉(zhuǎn)換操作，使本來僅需一次Shuffle的過程變?yōu)榱藞?zhí)行多次。我們就曾經(jīng)犯過這樣一個(gè)錯(cuò)誤,本來可以通過一次groupByKey完成的操作卻使用了兩回。業(yè)務(wù)邏輯是這樣的：我們有三張表分別是銷量（s）、價(jià)格（p）、庫存（v），每張表有3個(gè)字段：商品id（sku_id）、品類id（category）和歷史時(shí)序數(shù)據(jù)（data），現(xiàn)在需要按sku_id將s、p、v數(shù)據(jù)合并，然后再按category再合并一次，最終的數(shù)據(jù)格式是：[category，[[sku_id, s , p, v], [sku_id, s , p, v], […]，[…]]]。一開始我們先按照sku_id + category作為key進(jìn)行一次groupByKey，將數(shù)據(jù)格式轉(zhuǎn)換成[sku_id, category , [s，p, v]]，然后按category作為key再groupByKey一次。后來我們修改為按照category作為key只進(jìn)行一次groupByKey，因?yàn)橐粋€(gè)sku_id只會(huì)屬于一個(gè)category，所以后續(xù)的map轉(zhuǎn)換里面只需要寫一些代碼將相同sku_id的s、p、v數(shù)據(jù)group到一起就可以了。兩次groupByKey的情況：

修改后變?yōu)橐淮蝕roupByKey的情況：

多表join時(shí)，如果key值相同，則可以使用union+groupByKey+flatMapValues形式進(jìn)行。比如：需要將銷量、庫存、價(jià)格、促銷計(jì)劃和商品信息通過商品編碼連接到一起，一開始使用的是join轉(zhuǎn)換操作，將幾個(gè)RDD彼此join在一起。后來發(fā)現(xiàn)這樣做運(yùn)行速度非常慢，于是換成union+groypByKey+flatMapValue形式，這樣做只需進(jìn)行一次Shuffle，這樣修改后運(yùn)行速度比以前快多了。實(shí)例代碼如下：

如果兩個(gè)RDD需要在groupByKey后進(jìn)行join操作，可以使用cogroup轉(zhuǎn)換操作代替。比如， 將歷史銷量數(shù)據(jù)按品類進(jìn)行合并，然后再與模型文件進(jìn)行join操作，流程如下：

使用cogroup后，經(jīng)過一次Shuffle就可完成了兩步操作，性能大幅提升。

2. 減少Shuffle時(shí)傳輸?shù)臄?shù)據(jù)量

在Shuffle操作前盡量將不需要的數(shù)據(jù)過濾掉。

使用comebineyeByKey可以高效率的實(shí)現(xiàn)任何復(fù)雜的聚合邏輯。

comebineyeByKey屬于聚合類操作，由于它支持map端的聚合所以比groupByKey性能好，又由于它的map端與reduce端可以設(shè)置成不一樣的邏輯，所以它支持的場景比reduceByKey多，它的定義如下：

reduceByKey和groupByKey內(nèi)部實(shí)際是調(diào)用了comebineyeByKey，

我們之前有很多復(fù)雜的無法用reduceByKey來實(shí)現(xiàn)的聚合邏輯都通過groupByKey來完成的，后來全部替換為comebineyeByKey后性能提升了不少。

3. 處理數(shù)據(jù)傾斜

有些時(shí)候經(jīng)過一系列轉(zhuǎn)換操作之后數(shù)據(jù)變得十分傾斜，在這樣情況下后續(xù)的RDD計(jì)算效率會(huì)非常的糟糕，嚴(yán)重時(shí)程序報(bào)錯(cuò)。遇到這種情況通常會(huì)使用repartition這個(gè)轉(zhuǎn)換操作對RDD進(jìn)行重新分區(qū)，重新分區(qū)后數(shù)據(jù)會(huì)均勻分布在不同的分區(qū)中，避免了數(shù)據(jù)傾斜。如果是減少分區(qū)使用coalesce也可以達(dá)到效果，但比起repartition不足的是分配不是那么均勻。

 5. 小結(jié)

雖然京東的預(yù)測系統(tǒng)已經(jīng)穩(wěn)定運(yùn)行了很長一段時(shí)間，但是我們也看到系統(tǒng)本身還存在著很多待改進(jìn)的地方，接下來我們會(huì)在預(yù)測準(zhǔn)確度的提高、系統(tǒng)性能的優(yōu)化、多業(yè)務(wù)支持的便捷性上進(jìn)行改進(jìn)。未來，隨著大數(shù)據(jù)、人工智能技術(shù)在京東供應(yīng)鏈管理中的使用越來越多，預(yù)測系統(tǒng)也將發(fā)揮出更大作用，對于京東預(yù)測系統(tǒng)的研發(fā)工作也將是充滿著挑戰(zhàn)與樂趣。