背景介紹

前段時間京東公開了面向第二個十二年的戰(zhàn)略規(guī)劃，表示京東將全面走向技術(shù)化，大力發(fā)展人工智能和機器人自動化技術(shù)，將過去傳統(tǒng)方式構(gòu)筑的優(yōu)勢全面升級。京東Y事業(yè)部順勢成立，該事業(yè)部將以服務(wù)泛零售為核心，著重智能供應(yīng)能力的打造，核心使命是利用人工智能技術(shù)來驅(qū)動零售革新。

京東的供應(yīng)鏈

目前京東在全國范圍內(nèi)的運營256個大型倉庫，按功能可劃分為RDC、FDC、大件中心倉、大件衛(wèi)星倉、圖書倉和城市倉等等。RDC（Regional Distribution Center）即區(qū)域分發(fā)中心，可理解為一級倉庫，向供貨商采購的商品會優(yōu)先送往這里，一般設(shè)置在中心城市，覆蓋范圍大。FDC（Forward Distribution Center）即區(qū)域運轉(zhuǎn)中心，可理解為二級倉庫，覆蓋一些中、小型城市及邊遠(yuǎn)地區(qū)，通常會根據(jù)需求將商品從RDC調(diào)配過來。

結(jié)合人工智能、大數(shù)據(jù)等技術(shù)，京東首先從供貨商那里合理采購定量的商品到RDC，再根據(jù)實際需求調(diào)配到FDC，然后運往離客戶最近的配送站，最后快遞員將商品帶到客戶手中。這只是京東供應(yīng)鏈體系中一個普通的場景，但正因為有這樣的體系，使得京東對用戶的響應(yīng)速度大大提高，用戶體驗大大提升。

京東供應(yīng)鏈優(yōu)化

用戶體驗提升的同時也伴隨著大量資金的投入和成本的提高，成本必須得到控制，整個體系才能發(fā)揮出最大的價值，于是對供應(yīng)鏈的優(yōu)化就顯得至關(guān)重要了。

優(yōu)化其實是一門運籌學(xué)問題，需考慮在各種決策目標(biāo)之間如何平衡以達(dá)到最大收益，在這個過程中需要考慮很多問題，把這些考慮清楚，問題就容易解決了。舉幾個簡單的例子：

商品補貨：考慮在什么時間，給哪個RDC采購什么商品，采購量是多少？

商品調(diào)撥：考慮在什么時間，給哪個FDC調(diào)配什么商品，調(diào)配量是多少？

倉儲運營：在大促來臨之際，倉庫和配送站要增配多少人手、多少輛貨車？

雖然看上去這些問題都很容易回答，但仔細(xì)想想?yún)s又很難給出答案，原因就在于想要做到精確不是那么容易的事情，就拿補貨來說，補的太多會增加庫存成本，補的太少會增加缺貨成本，只有合理的補貨量才能做到成本最低。

預(yù)測技術(shù)在京東供應(yīng)鏈的作用

借助機器學(xué)習(xí)、大數(shù)據(jù)等相關(guān)技術(shù)，京東在很多供應(yīng)鏈優(yōu)化問題上都已經(jīng)實現(xiàn)系統(tǒng)化，由系統(tǒng)自動給出優(yōu)化建議，并與生產(chǎn)系統(tǒng)相連接，實現(xiàn)全流程自動化。在這里有一項技術(shù)起著至關(guān)重要的低層支撐作用——預(yù)測技術(shù)。據(jù)粗略估算，1%的預(yù)測準(zhǔn)確度的提升可以節(jié)約數(shù)倍的運營成本。

怎樣理解預(yù)測在供應(yīng)鏈優(yōu)化中的作用呢?拿商品補貨舉例，一家公司為了保證庫房不缺貨，可能會頻繁的從供貨商那里補充大量商品，這樣做雖然不會缺貨，但可能會造成更多賣不出去的商品積壓在倉庫中，從而使商品的周轉(zhuǎn)率降低，庫存成本增加。反之，這家公司有可能為了追求零庫存而補很少的商品，但這就可能出現(xiàn)嚴(yán)重的缺貨問題，從而使現(xiàn)貨率降低，嚴(yán)重影響用戶體驗，缺貨成本增加。于是問題就來了，要補多少商品才合適，什么時間補貨，這就需要權(quán)衡考慮了，最終目的是要使庫存成本和缺貨成本達(dá)到一個平衡。

考慮一下極端情況，等庫存降到零時再去補貨，這時供貨商接到補貨通知后將貨物運往倉庫。但是這么做有個問題，因為運送過程需要時間，這段時間庫房就缺貨了。那怎么辦呢?就是利用預(yù)測技術(shù)。利用預(yù)測我們可以計算出未來商品在途的這段時間里銷量大概是多少，然后我們讓倉庫保證這個量，低于這個量就給供貨商下達(dá)補貨通知，于是問題得以解決。總而言之，預(yù)測技術(shù)在這里發(fā)揮了重要的作用，成為關(guān)鍵的一個環(huán)。

京東預(yù)測系統(tǒng)

預(yù)測系統(tǒng)介紹

預(yù)測系統(tǒng)在整個供應(yīng)鏈體系中處在最底層并且起到一個支撐的作用，支持上層的多個決策優(yōu)化系統(tǒng)，而這些決策優(yōu)化系統(tǒng)利用精準(zhǔn)的預(yù)測數(shù)據(jù)結(jié)合運籌學(xué)技術(shù)得出最優(yōu)的決策，并將結(jié)果提供給更上層的業(yè)務(wù)執(zhí)行系統(tǒng)或是業(yè)務(wù)方直接使用。

目前，預(yù)測系統(tǒng)主要支持三大業(yè)務(wù)：銷量預(yù)測、單量預(yù)測和GMV預(yù)測。其中銷量預(yù)測主要支持商品補貨、商品調(diào)撥；單量預(yù)測主要支持倉庫、站點的運營管理；GMV預(yù)測主要支持銷售部門計劃的定制。

銷量預(yù)測按照不同維度又可以分為RDC采購預(yù)測、FDC調(diào)撥預(yù)測、城市倉調(diào)撥預(yù)測、大建倉補貨預(yù)測、全球購銷量預(yù)測和圖書促銷預(yù)測等；單量預(yù)測又可分為庫房單量預(yù)測、配送中心單量預(yù)測和配送站單量預(yù)測等（在這里“單量”并非指用戶所下訂單的量，而是將訂單拆單后流轉(zhuǎn)到倉庫中的單量。例如一個用戶的訂單中包括3件物品，其中兩個大件品和一個小件品，在京東的供應(yīng)鏈環(huán)節(jié)中可能會將其中兩個大件品組成一個單投放到大件倉中，而將那個小件單獨一個單投放到小件倉中，單量指的是拆單后的量）；GMV預(yù)測支持到商品粒度。

預(yù)測系統(tǒng)架構(gòu)

整體架構(gòu)從上至下依次是：數(shù)據(jù)源輸入層、基礎(chǔ)數(shù)據(jù)加工層、核心業(yè)務(wù)層、數(shù)據(jù)輸出層和下游系統(tǒng)。首先從外部數(shù)據(jù)源獲取我們所需的業(yè)務(wù)數(shù)據(jù)，然后對基礎(chǔ)數(shù)據(jù)進(jìn)行加工清洗，再通過時間序列、機器學(xué)習(xí)等人工智能技術(shù)對數(shù)據(jù)進(jìn)行處理分析，最后計算出預(yù)測結(jié)果并通過多種途徑推送給下游系統(tǒng)使用。

數(shù)據(jù)源輸入層：京東數(shù)據(jù)倉庫中存儲著我們需要的大部分業(yè)務(wù)數(shù)據(jù)，例如訂單信息、商品信息、庫存信息等等。而對于促銷計劃數(shù)據(jù)則大部分來自于采銷人員通過Web系統(tǒng)錄入的信息。除此之外還有一小部分?jǐn)?shù)據(jù)通過文本形式直接上傳到HDFS中。

基礎(chǔ)數(shù)據(jù)加工層：在這一層主要通過Hive對基礎(chǔ)數(shù)據(jù)進(jìn)行一些加工清洗，去掉不需要的字段，過濾不需要的維度并清洗有問題的數(shù)據(jù)。

核心業(yè)務(wù)層：這層是系統(tǒng)的的核心部分，橫向看又可分為三層：特征構(gòu)建、預(yù)測算法和預(yù)測結(jié)果加工。縱向看是由多條業(yè)務(wù)線組成，彼此之間不發(fā)生任何交集。

• 特征構(gòu)建：將之前清洗過的基礎(chǔ)數(shù)據(jù)通過近一步的處理轉(zhuǎn)化成標(biāo)準(zhǔn)格式的特征數(shù)據(jù)，提供給后續(xù)算法模型使用。
• 核心算法：利用時間序列分析、機器學(xué)習(xí)等人工智能技術(shù)進(jìn)行銷量、單量的預(yù)測，是預(yù)測系統(tǒng)中最為核心的部分。
• 預(yù)測結(jié)果加工：預(yù)測結(jié)果可能在格式和一些特殊性要求上不能滿足下游系統(tǒng)，所以還需要根據(jù)實際情況對其進(jìn)行加工處理，比如增加標(biāo)準(zhǔn)差、促銷標(biāo)識等額外信息。

預(yù)測結(jié)果輸出層：將最終預(yù)測結(jié)果同步回京東數(shù)據(jù)倉庫、MySql、HBase或制作成JSF接口供其他系統(tǒng)遠(yuǎn)程調(diào)用。

下游系統(tǒng)：包括下游任務(wù)流程、下游Web系統(tǒng)和其他系統(tǒng)。

預(yù)測系統(tǒng)核心介紹

預(yù)測系統(tǒng)核心層技術(shù)選型 預(yù)測系統(tǒng)核心層技術(shù)主要分為四層：基礎(chǔ)層、框架層、工具層和算法層。

基礎(chǔ)層：

HDFS用來做數(shù)據(jù)存儲，Yarn用來做資源調(diào)度，BDP（Big Data Platform）是京東自己研發(fā)的大數(shù)據(jù)平臺，我們主要用它來做任務(wù)調(diào)度。

框架層：

以Spark RDD、Spark SQL、Hive為主， MapReduce程序占一小部分，是原先遺留下來的，目前正逐步替換成Spark RDD。 選擇Spark除了對性能的考慮外，還考慮了Spark程序開發(fā)的高效率、多語言特性以及對機器學(xué)習(xí)算法的支持。在Spark開發(fā)語言上我們選擇了Python，原因有以下三點：

• Python有很多不錯的機器學(xué)習(xí)算法包可以使用，比起Spark的MLlib，算法的準(zhǔn)確度更高。我們用GBDT做過對比，發(fā)現(xiàn)xgboost比MLlib里面提供的提升樹模型預(yù)測準(zhǔn)確度高出大概5%~10%。雖然直接使用Spark自帶的機器學(xué)習(xí)框架會節(jié)省我們的開發(fā)成本，但預(yù)測準(zhǔn)確度對于我們來說至關(guān)重要，每提升1%的準(zhǔn)確度，就可能會帶來成本的成倍降低。
• 我們的團隊中包括開發(fā)工程師和算法工程師，對于算法工程師而言他們更擅長使用Python進(jìn)行數(shù)據(jù)分析，使用Java或Scala會有不小的學(xué)習(xí)成本。
• 對比其他語言，我們發(fā)現(xiàn)使用Python的開發(fā)效率是最高的，并且對于一個新人，學(xué)習(xí)Python比學(xué)習(xí)其他語言更加容易。

工具層：

一方面我們會結(jié)合自身業(yè)務(wù)有針對性的開發(fā)一些算法，另一方面我們會直接使用業(yè)界比較成熟的算法和模型，這些算法都封裝在第三方Python包中。我們比較常用的包有xgboost、numpy、pandas、sklearn、scipy和hyperopt等。

• Xgboost：它是Gradient Boosting Machine的一個C++實現(xiàn)，xgboost最大的特點在于，它能夠自動利用CPU的多線程進(jìn)行并行，同時在算法上加以改進(jìn)提高了精度。
• numpy：是Python的一種開源的數(shù)值計算擴展。這種工具可用來存儲和處理大型矩陣，比Python自身的嵌套列表結(jié)構(gòu)要高效的多（該結(jié)構(gòu)也可以用來表示矩陣）。
• pandas：是基于NumPy 的一種工具，該工具是為了解決數(shù)據(jù)分析任務(wù)而創(chuàng)建的。Pandas 納入了大量庫和一些標(biāo)準(zhǔn)的數(shù)據(jù)模型，提供了高效地操作大型數(shù)據(jù)集所需的工具。
• sklearn：是Python重要的機器學(xué)習(xí)庫，支持包括分類、回歸、降維和聚類四大機器學(xué)習(xí)算法。還包含了特征提取、數(shù)據(jù)處理和模型評估三大模塊。
• scipy：是在NumPy庫的基礎(chǔ)上增加了眾多的數(shù)學(xué)、科學(xué)以及工程計算中常用的庫函數(shù)。例如線性代數(shù)、常微分方程數(shù)值求解、信號處理、圖像處理和稀疏矩陣等等。

算法層：

我們用到的算法模型非常多，原因是京東的商品品類齊全、業(yè)務(wù)復(fù)雜，需要根據(jù)不同的情況采用不同的算法模型。我們有一個獨立的系統(tǒng)來為算法模型與商品之間建立匹配關(guān)系，有些比較復(fù)雜的預(yù)測業(yè)務(wù)還需要使用多個模型。我們使用的算法總體上可以分為三類：時間序列、機器學(xué)習(xí)和結(jié)合業(yè)務(wù)開發(fā)的一些獨有的算法。

1. 機器學(xué)習(xí)算法主要包括GBDT、LASSO和RNN ：

GBDT：是一種迭代的決策樹算法，該算法由多棵決策樹組成，所有樹的結(jié)論累加起來做最終答案。我們用它來預(yù)測高銷量，但歷史規(guī)律不明顯的商品。

RNN：這種網(wǎng)絡(luò)的內(nèi)部狀態(tài)可以展示動態(tài)時序行為。不同于前饋神經(jīng)網(wǎng)絡(luò)的是，RNN可以利用它內(nèi)部的記憶來處理任意時序的輸入序列，這讓它可以更容易處理如時序預(yù)測、語音識別等。

LASSO：該方法是一種壓縮估計。它通過構(gòu)造一個罰函數(shù)得到一個較為精煉的模型，使得它壓縮一些系數(shù)，同時設(shè)定一些系數(shù)為零。因此保留了子集收縮的優(yōu)點，是一種處理具有復(fù)共線性數(shù)據(jù)的有偏估計。用來預(yù)測低銷量，歷史數(shù)據(jù)平穩(wěn)的商品效果較好。

2. 時間序列主要包括ARIMA和Holt winters ：

ARIMA：全稱為自回歸積分滑動平均模型，于70年代初提出的一個著名時間序列預(yù)測方法，我們用它來主要預(yù)測類似庫房單量這種平穩(wěn)的序列。

Holt winters：又稱三次指數(shù)平滑算法，也是一個經(jīng)典的時間序列算法，我們用它來預(yù)測季節(jié)性和趨勢都很明顯的商品。

3. 結(jié)合業(yè)務(wù)開發(fā)的獨有算法包括WMAStockDT、SimilarityModel和NewProduct等：

WMAStockDT：庫存決策樹模型，用來預(yù)測受庫存狀態(tài)影響較大的商品。

SimilarityModel：相似品模型，使用指定的同類品數(shù)據(jù)來預(yù)測某商品未來銷量。

NewProduct：新品模型，顧名思義就是用來預(yù)測新品的銷量。

預(yù)測系統(tǒng)核心流程

預(yù)測核心流程主要包括兩類：以機器學(xué)習(xí)算法為主的流程和以時間序列分析為主的流程。

1. 以機器學(xué)習(xí)算法為主的流程如下：

• 特征構(gòu)建：通過數(shù)據(jù)分析、模型試驗確定主要特征，通過一系列任務(wù)生成標(biāo)準(zhǔn)格式的特征數(shù)據(jù)。
• 模型選擇：不同的商品有不同的特性，所以首先會根據(jù)商品的銷量高低、新品舊品、假節(jié)日敏感性等因素分配不同的算法模型。
• 特征選擇：對一批特征進(jìn)行篩選過濾不需要的特征，不同類型的商品特征不同。
• 樣本分區(qū)：對訓(xùn)練數(shù)據(jù)進(jìn)行分組，分成多組樣本，真正訓(xùn)練時針對每組樣本生成一個模型文件。一般是同類型商品被分成一組，比如按品類維度分組，這樣做是考慮并行化以及模型的準(zhǔn)確性。
• 模型參數(shù)：選擇最優(yōu)的模型參數(shù)，合適的參數(shù)將提高模型的準(zhǔn)確度，因為需要對不同的參數(shù)組合分別進(jìn)行模型訓(xùn)練和預(yù)測，所以這一步是非常耗費資源。
• 模型訓(xùn)練：待特征、模型、樣本都確定好后就可以進(jìn)行模型訓(xùn)練，訓(xùn)練往往會耗費很長時間，訓(xùn)練后會生成模型文件，存儲在HDFS中。
• 模型預(yù)測：讀取模型文件進(jìn)行預(yù)測執(zhí)行。
• 多模型擇優(yōu)：為了提高預(yù)測準(zhǔn)確度，我們可能會使用多個算法模型，當(dāng)每個模型的預(yù)測結(jié)果輸出后系統(tǒng)會通過一些規(guī)則來選擇一個最優(yōu)的預(yù)測結(jié)果。
• 預(yù)測值異常攔截：我們發(fā)現(xiàn)越是復(fù)雜且不易解釋的算法越容易出現(xiàn)極個別預(yù)測值異常偏高的情況，這種預(yù)測偏高無法結(jié)合歷史數(shù)據(jù)進(jìn)行解釋，因此我們會通過一些規(guī)則將這些異常值攔截下來，并且用一個更加保守的數(shù)值代替。
• 模型評價：計算預(yù)測準(zhǔn)確度，我們通常用使用mapd來作為評價指標(biāo)。
• 誤差分析：通過分析預(yù)測準(zhǔn)確度得出一個誤差在不同維度上的分布，以便給算法優(yōu)化提供參考依據(jù)。

2. 以時間序列分析為主的預(yù)測流程如下：

• 生成歷史時序：將歷史銷量、價格、庫存等數(shù)據(jù)按照規(guī)定格式生成時序數(shù)據(jù)。
• 節(jié)假日因子：計算節(jié)假日與銷量之間的關(guān)系，用來平滑節(jié)假日對銷量影響。
• 周日因子：計算周一到周日這7天與銷量的關(guān)系，用來平滑周日對銷量的影響。
• 促銷因子：計算促銷與銷量之間的關(guān)系，用來平滑促銷對銷量的影響。
• 因子平滑：歷史銷量是不穩(wěn)定的，會受到節(jié)假日、促銷等影響，在這種情況下進(jìn)行預(yù)測有很大難度，所以需要利用之前計算的各類因子對歷史數(shù)據(jù)進(jìn)行平滑處理。
• 時序預(yù)測：在一個相對平穩(wěn)的銷量數(shù)據(jù)上通過算法進(jìn)行預(yù)測。
• 因子疊加：結(jié)合未來節(jié)假日、促銷計劃等因素對預(yù)測結(jié)果進(jìn)行調(diào)整。

Spark在預(yù)測核心層的應(yīng)用

我們使用Spark SQL和Spark RDD相結(jié)合的方式來編寫程序，對于一般的數(shù)據(jù)處理，我們使用Spark的方式與其他無異，但是對于模型訓(xùn)練、預(yù)測這些需要調(diào)用算法接口的邏輯就需要考慮一下并行化的問題了。我們平均一個訓(xùn)練任務(wù)在一天處理的數(shù)據(jù)量大約在500G左右，雖然數(shù)據(jù)規(guī)模不是特別的龐大，但是Python算法包提供的算法都是單進(jìn)程執(zhí)行。我們計算過，如果使用一臺機器訓(xùn)練全部品類數(shù)據(jù)需要一個星期的時間，這是無法接收的，所以我們需要借助Spark這種分布式并行計算框架來將計算分?jǐn)偟蕉鄠€節(jié)點上實現(xiàn)并行化處理。

我們實現(xiàn)的方法很簡單，首先需要在集群的每個節(jié)點上安裝所需的全部Python包，然后在編寫Spark程序時考慮通過某種規(guī)則將數(shù)據(jù)分區(qū)，比如按品類維度，通過groupByKey操作將數(shù)據(jù)重新分區(qū)，每一個分區(qū)是一個樣本集合并進(jìn)行獨立的訓(xùn)練，以此達(dá)到并行化。流程如下圖所示：

偽碼如下：

repartitionBy方法即設(shè)置一個重分區(qū)的邏輯返回(K,V)結(jié)構(gòu)RDD，train方法是訓(xùn)練數(shù)據(jù)，在train方法里面會調(diào)用Python算法包接口。saveAsPickleFile是Spark Python獨有的一個Action操作，支持將RDD保存成序列化后的sequnceFile格式的文件，在序列化過程中會以10個一批的方式進(jìn)行處理，保存模型文件非常適合。

雖然原理簡單，但存在著一個難點，即以什么樣的規(guī)則進(jìn)行分區(qū)，key應(yīng)該如何設(shè)置。為了解決這個問題我們需要考慮幾個方面，第一就是哪些數(shù)據(jù)應(yīng)該被聚合到一起進(jìn)行訓(xùn)練，第二就是如何避免數(shù)據(jù)傾斜。

針對第一個問題我們做了如下幾點考慮：

• 被分在一個分區(qū)的數(shù)據(jù)要有一定的相似性，這樣訓(xùn)練的效果才會更好，比如按品類分區(qū)就是個典型例子。
• 分析商品的特性，根據(jù)特性的不同選擇不同的模型，例如高銷商品和低銷商品的預(yù)測模型是不一樣的，即使是同一模型使用的特征也可能不同，比如對促銷敏感的商品就需要更多與促銷相關(guān)特征，相同模型相同特征的商品應(yīng)傾向于分在一個分區(qū)中。

針對第二個問題我們采用了如下的方式解決：

• 對于數(shù)據(jù)量過大的分區(qū)進(jìn)行隨機抽樣選取。
• 對于數(shù)據(jù)量過大的分區(qū)還可以做二次拆分，比如圖書小說這個品類數(shù)據(jù)量明顯大于其他品類，于是就可以分析小說品類下的子品類數(shù)據(jù)量分布情況，并將子品類合并成新的幾個分區(qū)。
• 對于數(shù)據(jù)量過小這種情況則需要考慮進(jìn)行幾個分區(qū)數(shù)據(jù)的合并處理。

總之對于后兩種處理方式可以單獨通過一個Spark任務(wù)定期運行，并將這種分區(qū)規(guī)則保存。

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

注：《圖解Spark：核心技術(shù)與案例實戰(zhàn)》為本文作者所著。

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

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

在圖解Spark書的第六章描述了Spark運行架構(gòu)，介紹了Spark集群資源調(diào)度一般分為粗粒度調(diào)度和細(xì)粒度調(diào)度兩種模式。粗粒度包括了獨立運行模式和Mesos粗粒度運行模式，在這種情況下以整個機器作為分配單元執(zhí)行作業(yè)，該模式優(yōu)點是由于資源長期持有減少了資源調(diào)度的時間開銷，缺點是該模式中無法感知資源使用的變化，易造成系統(tǒng)資源的閑置，從而造成了資源浪費。

而細(xì)粒度包括了Yarn運行模式和Mesos細(xì)粒度運行模式,該模式的優(yōu)點是系統(tǒng)資源能夠得到充分利用，缺點是該模式中每個任務(wù)都需要從管理器獲取資源，調(diào)度延遲較大、開銷較大。

由于京東Spark集群屬于基礎(chǔ)平臺，在公司內(nèi)部共享這些資源，所以集群采用的是Yarn運行模式，在這種模式下可以根據(jù)不同系統(tǒng)所需要的資源進(jìn)行靈活的管理。在YARN-Cluster模式中，當(dāng)用戶向YARN集群中提交一個應(yīng)用程序后，YARN集群將分兩個階段運行該應(yīng)用程序：

第一個階段是把Spark的SparkContext作為Application Master在YARN集群中先啟動；第二個階段是由Application Master創(chuàng)建應(yīng)用程序，然后為它向Resource Manager申請資源，并啟動Executor來運行任務(wù)集，同時監(jiān)控它的整個運行過程，直到運行完成。下圖為Yarn-Cluster運行模式執(zhí)行過程：

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

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

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

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

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

修改設(shè)計

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

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

修改程序邏輯

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

1. 避免過多的Shuffle

Spark提供了豐富的轉(zhuǎn)換操作，可以使我們完成各類復(fù)雜的數(shù)據(jù)處理工作，但是也正因為如此我們在寫Spark程序的時候可能會遇到一個陷阱，那就是為了使代碼變的簡潔過分依賴RDD的轉(zhuǎn)換操作，使本來僅需一次Shuffle的過程變?yōu)榱藞?zhí)行多次。我們就曾經(jīng)犯過這樣一個錯誤,本來可以通過一次groupByKey完成的操作卻使用了兩回。

業(yè)務(wù)邏輯是這樣的：我們有三張表分別是銷量（s）、價格（p）、庫存（v），每張表有3個字段：商品id（sku_id）、品類id（category）和歷史時序數(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，因為一個sku_id只會屬于一個category，所以后續(xù)的map轉(zhuǎn)換里面只需要寫一些代碼將相同sku_id的s、p、v數(shù)據(jù)group到一起就可以了。兩次groupByKey的情況：

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

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

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

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

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

• 在Shuffle操作前盡量將不需要的數(shù)據(jù)過濾掉。
• 使用comebineyeByKey可以高效率的實現(xiàn)任何復(fù)雜的聚合邏輯。

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

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

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

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

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

寫在最后

雖然京東的預(yù)測系統(tǒng)已經(jīng)穩(wěn)定運行了很長一段時間，但是我們也看到系統(tǒng)本身還存在著很多待改進(jìn)的地方，接下來我們會在預(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)與樂趣。