4. 結合圖解Spark書進行應用與優化

《圖解Spark：核心技術與案例實戰》一書以Spark2.0版本為基礎進行編寫，系統介紹了Spark核心及其生態圈組件技術。其內容包括Spark生態圈、實戰環境搭建和編程模型等，重點介紹了作業調度、容錯執行、監控管理、存儲管理以及運行架構，同時還介紹了Spark生態圈相關組件，包括了Spark SQL的即席查詢、Spark Streaming的實時流處理、MLlib的機器學習、GraphX的圖處理和Alluxio的分布式內存文件系統等。下面介紹京東預測系統如何進行資源調度，并描述如何使用Spark存儲相關知識進行系統優化。

4.1 結合系統中的應用

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

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

 4.2 結合系統的優化

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

由于資源限制，分配給預測系統的Spark集群規模并不是很大,在有限的資源下運行Spark應用程序確實是一個考驗，因為在這種情況下經常會出現諸如程序計算時間太長、找不到Executor等錯誤。我們通過調整參數、修改設計和修改程序邏輯三個方面進行優化：

4.2.1 參數調整

• 減少num-executors，調大executor-memory，這樣的目的是希望Executor有足夠的內存可以使用。
• 查看日志發現沒有足夠的空間存儲廣播變量，分析是由于Cache到內存里的數據太多耗盡了內存，于是我們將Cache的級別適當調成MEMORY_ONLY_SER和DISK_ONLY。
• 針對某些任務關閉了推測機制，因為有些任務會出現暫時無法解決的數據傾斜問題，并非節點出現問題。
• 調整內存分配，對于一個Shuffle很多的任務，我們就把Cache的內存分配比例調低，同時調高Shuffle的內存比例。

 4.2.2 修改設計

參數的調整雖然容易做，但往往效果不好，這時候需要考慮從設計的角度去優化：

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

4.2.3 修改程序邏輯

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

1. 避免過多的Shuffle

Spark提供了豐富的轉換操作，可以使我們完成各類復雜的數據處理工作，但是也正因為如此我們在寫Spark程序的時候可能會遇到一個陷阱，那就是為了使代碼變的簡潔過分依賴RDD的轉換操作，使本來僅需一次Shuffle的過程變為了執行多次。我們就曾經犯過這樣一個錯誤,本來可以通過一次groupByKey完成的操作卻使用了兩回。業務邏輯是這樣的：我們有三張表分別是銷量（s）、價格（p）、庫存（v），每張表有3個字段：商品id（sku_id）、品類id（category）和歷史時序數據（data），現在需要按sku_id將s、p、v數據合并，然后再按category再合并一次，最終的數據格式是：[category，[[sku_id, s , p, v], [sku_id, s , p, v], […]，[…]]]。一開始我們先按照sku_id + category作為key進行一次groupByKey，將數據格式轉換成[sku_id, category , [s，p, v]]，然后按category作為key再groupByKey一次。后來我們修改為按照category作為key只進行一次groupByKey，因為一個sku_id只會屬于一個category，所以后續的map轉換里面只需要寫一些代碼將相同sku_id的s、p、v數據group到一起就可以了。兩次groupByKey的情況：

修改后變為一次groupByKey的情況：

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

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

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

2. 減少Shuffle時傳輸的數據量

在Shuffle操作前盡量將不需要的數據過濾掉。

使用comebineyeByKey可以高效率的實現任何復雜的聚合邏輯。

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

reduceByKey和groupByKey內部實際是調用了comebineyeByKey，

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

3. 處理數據傾斜

有些時候經過一系列轉換操作之后數據變得十分傾斜，在這樣情況下后續的RDD計算效率會非常的糟糕，嚴重時程序報錯。遇到這種情況通常會使用repartition這個轉換操作對RDD進行重新分區，重新分區后數據會均勻分布在不同的分區中，避免了數據傾斜。如果是減少分區使用coalesce也可以達到效果，但比起repartition不足的是分配不是那么均勻。

 5. 小結

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