作者：Yanjun

Spark Application可以直接運行在YARN集群上，這種運行模式，會將資源的管理與協(xié)調(diào)統(tǒng)一交給YARN集群去處理，這樣能夠?qū)崿F(xiàn)構建于YARN集群之上Application的多樣性，比如可以運行MapReduc程序，可以運行HBase集群，也可以運行Storm集群，還可以運行使用Python開發(fā)機器學習應用程序，等等。

我們知道，Spark on YARN又分為client模式和cluster模式：在client模式下，Spark Application運行的Driver會在提交程序的節(jié)點上，而該節(jié)點可能是YARN集群內(nèi)部節(jié)點，也可能不是，一般來說提交Spark Application的客戶端節(jié)點不是YARN集群內(nèi)部的節(jié)點，那么在客戶端節(jié)點上可以根據(jù)自己的需要安裝各種需要的軟件和環(huán)境，以支撐Spark Application正常運行。在cluster模式下，Spark Application運行時的所有進程都在YARN集群的NodeManager節(jié)點上，而且具體在哪些NodeManager上運行是由YARN的調(diào)度策略所決定的。

對比這兩種模式，最關鍵的是Spark Application運行時Driver所在的節(jié)點不同，而且，如果想要對Driver所在節(jié)點的運行環(huán)境進行配置，區(qū)別很大，但這對于PySpark Application運行來說是非常關鍵的。

PySpark是Spark為使用Python程序編寫Spark Application而實現(xiàn)的客戶端庫，通過PySpark也可以編寫Spark Application并在Spark集群上運行。Python具有非常豐富的科學計算、機器學習處理庫，如numpy、pandas、scipy等等。為了能夠充分利用這些高效的Python模塊，很多機器學習程序都會使用Python實現(xiàn)，同時也希望能夠在Spark集群上運行。

PySpark Application運行原理

理解PySpark Application的運行原理，有助于我們使用Python編寫Spark Application，并能夠?qū)ySpark Application進行各種調(diào)優(yōu)。PySpark構建于Spark的Java API之上，數(shù)據(jù)在Python腳本里面進行處理，而在JVM中緩存和Shuffle數(shù)據(jù)，數(shù)據(jù)處理流程如下圖所示（來自Apache Spark Wiki）：

Spark Application會在Driver中創(chuàng)建pyspark.SparkContext對象，后續(xù)通過pyspark.SparkContext對象來構建Job DAG并提交DAG運行。使用Python編寫PySpark Application，在Python編寫的Driver中也有一個pyspark.SparkContext對象，該pyspark.SparkContext對象會通過Py4J模塊啟動一個JVM實例，創(chuàng)建一個JavaSparkContext對象。PY4J只用在Driver上，后續(xù)在Python程序與JavaSparkContext對象之間的通信，都會通過PY4J模塊來實現(xiàn)，而且都是本地通信。

PySpark Application中也有RDD，對Python RDD的Transformation操作，都會被映射到Java中的PythonRDD對象上。對于遠程節(jié)點上的Python RDD操作，Java PythonRDD對象會創(chuàng)建一個Python子進程，并基于Pipe的方式與該Python子進程通信，將用戶編寫Python處理代碼和數(shù)據(jù)發(fā)送到Python子進程中進行處理。

下面，我們基于Spark on YARN模式，并根據(jù)當前企業(yè)所具有的實際集群運行環(huán)境情況，來說明如何在Spark集群上運行PySpark Application，大致分為如下3種情況：

這種情況，如果是初始安裝YARN、Spark集群，并考慮到了當前應用場景需要支持Python程序運行在Spark集群之上，這時可以準備好對應Python軟件包、依賴模塊，在YARN集群中的每個節(jié)點上進行安裝。這樣，YARN集群的每個NodeManager上都具有Python環(huán)境，可以編寫PySpark Application并在集群上運行。目前比較流行的是直接安裝Python虛擬環(huán)境，使用Anaconda等軟件，可以極大地簡化Python環(huán)境的管理工作。

這種方式的缺點是，如果后續(xù)使用Python編寫Spark Application，需要增加新的依賴模塊，那么就需要在YARN集群的每個節(jié)點上都進行該新增模塊的安裝。而且，如果依賴Python的版本，可能還需要管理不同版本Python環(huán)境。因為提交PySpark Application運行，具體在哪些NodeManager上運行該Application，是由YARN的調(diào)度器決定的，必須保證每個NodeManager上都具有Python環(huán)境（基礎環(huán)境+依賴模塊）。

這種情況，更適合企業(yè)已經(jīng)安裝了規(guī)模較大的YARN集群，并在開始使用時并未考慮到后續(xù)會使用基于Python來編寫Spark Application，并且不想在YARN集群的NodeManager上安裝Python環(huán)境依賴依賴模塊。我們參考了Benjamin Zaitlen的博文（詳見后面參考鏈接），并基于Anaconda軟件環(huán)境進行了實踐和驗證，具體實現(xiàn)思路如下所示：

下面進行詳細說明：

首先，我們在CentOS 7.2上，基于Python 2.7，下載了Anaconda2-5.0.0.1-Linux-x86_64.sh安裝軟件，并進行了安裝。Anaconda的安裝路徑為/root/anaconda2。

然后，創(chuàng)建一個Python虛擬環(huán)境，執(zhí)行如下命令：

conda create -n mlpy_env --copy -y -q python=2 numpy pandas scipy

上述命令創(chuàng)建了一個名稱為mlpy_env的Python環(huán)境，–copy選項將對應的軟件包都安裝到該環(huán)境中，包括一些C的動態(tài)鏈接庫文件。同時，下載numpy、pandas、scipy這三個依賴模塊到該環(huán)境中。

接著，將該Python環(huán)境打包，執(zhí)行如下命令：

cd?/root/anaconda2/envszip -r mlpy_env.zip mlpy_env

該zip文件大概有400MB左右，將該zip壓縮包拷貝到指定目錄中，方便后續(xù)提交PySpark Application：

最后，我們可以提交我們的PySpark Application，執(zhí)行如下命令：

1PYSPARK_PYTHON=./ANACONDA/mlpy_env/bin/python spark-submit \2?--conf spark.yarn.appMasterEnv.PYSPARK_PYTHON=./ANACONDA/mlpy_env/bin/python \3?--master yarn-cluster \4?--archives /tmp/mlpy_env.zip#ANACONDA \5?/var/lib/hadoop-hdfs/pyspark/test_pyspark_dependencies.py

上面的test_pyspark_dependencies.py文件中，使用了numpy、pandas、scipy這三個依賴包的函數(shù)，通過上面提到的YARN集群的cluster模式可以運行在Spark集群上。

可以看到，上面的依賴zip壓縮包將整個Python的運行環(huán)境都包含在里面，在提交PySpark Application時會將該環(huán)境zip包上傳到運行Application的所在的每個節(jié)點上，并解壓縮后為Python代碼提供運行時環(huán)境。如果不想每次都從客戶端將該環(huán)境文件上傳到集群中運行PySpark Application的節(jié)點上，也可以將zip包上傳到HDFS上，并修改–archives參數(shù)的值為hdfs:///tmp/mlpy_env.zip#ANACONDA，也是可以的。

另外，需要說明的是，如果我們開發(fā)的/var/lib/hadoop-hdfs/pyspark/test_pyspark_dependencies.py文件中，也依賴的一些我們自己實現(xiàn)的處理函數(shù)，具有多個Python依賴的文件，想要通過上面的方式運行，必須將這些依賴的Python文件拷貝到我們創(chuàng)建的環(huán)境中，對應的目錄為mlpy_env/lib/python2.7/site-packages/下面。

假如我們還是希望使用Spark on YARN模式來運行PySpark Application，但并不將Python程序提交到YARN集群上運行。這時，我們可以考慮使用混合編程語言的方式，來處理數(shù)據(jù)任務。比如，機器學習Application具有迭代計算的特性，更適合在一個高配的節(jié)點上運行；而普通的ETL數(shù)據(jù)處理具有多機并行處理的特點，適合放到集群上進行分布式處理。

一個完整的機器學習Application的設計與構建，可以將算法部分和數(shù)據(jù)準備部分分離出來，使用Scala/Java進行數(shù)據(jù)預處理，輸出一個機器學習算法所需要（更便于迭代、尋優(yōu)計算）的輸入數(shù)據(jù)格式，這會極大地壓縮算法輸入數(shù)據(jù)的規(guī)模，從而使算法迭代計算充分利用單機本地的資源（內(nèi)存、CPU、網(wǎng)絡），這可能會比直接放到集群中計算要快得多。

因此，我們在對機器學習Application準備數(shù)據(jù)時，使用原生的Scala編程語言實現(xiàn)Spark Application來處理數(shù)據(jù)，包括轉換、統(tǒng)計、壓縮等等，將滿足算法輸入格式的數(shù)據(jù)輸出到HDFS指定目錄中。在性能方面，對數(shù)據(jù)規(guī)模較大的情況下，在Spark集群上處理數(shù)據(jù)，Scala/Java實現(xiàn)的Spark Application運行性能要好一些。然后，算法迭代部分，基于豐富、高性能的Python科學計算模塊，使用Python語言實現(xiàn)，其實直接使用PySpark API實現(xiàn)一個機器學習PySpark Application，運行模式為YARN client模式。這時，就需要在算法運行的節(jié)點上安裝好Python環(huán)境及其依賴模塊（而不需要在YARN集群的節(jié)點上安裝），Driver程序從HDFS中讀取輸入數(shù)據(jù)（緩存到本地），然后在本地進行算法的迭代計算，最后輸出模型。

總結

對于重度使用PySpark的情況，比如偏向機器學習，可以考慮在整個集群中都安裝好Python環(huán)境，并根據(jù)不同的需要進行依賴模塊的統(tǒng)一管理，能夠=極大地方便PySpark Application的運行。

不在YARN集群上安裝Python環(huán)境的方案，會使提交的Python環(huán)境zip包在YARN集群中傳輸帶來一定開銷，而且每次提交一個PySpark Application都需要打包一個環(huán)境zip文件，如果有大量的Python實現(xiàn)的PySpark Application需要在Spark集群上運行，開銷會越來越大。另外，如果PySpark應用程序修改，可能需要重新打包環(huán)境。但是這樣做確實不在需要考慮YARN集群集群節(jié)點上的Python環(huán)境了，任何版本Python編寫的PySpark Application都可以使用集群資源運行。

關于該問題，SPARK-13587（詳見下面參考鏈接）也在討論如果優(yōu)化該問題，后續(xù)應該會有一個比較合適的解決方案。

生成海報

免責聲明：本網(wǎng)站內(nèi)容主要來自原創(chuàng)、合作伙伴供稿和第三方自媒體作者投稿，凡在本網(wǎng)站出現(xiàn)的信息，均僅供參考。本網(wǎng)站將盡力確保所提供信息的準確性及可靠性，但不保證有關資料的準確性及可靠性，讀者在使用前請進一步核實，并對任何自主決定的行為負責。本網(wǎng)站對有關資料所引致的錯誤、不確或遺漏，概不負任何法律責任。任何單位或個人認為本網(wǎng)站中的網(wǎng)頁或鏈接內(nèi)容可能涉嫌侵犯其知識產(chǎn)權或存在不實內(nèi)容時，應及時向本網(wǎng)站提出書面權利通知或不實情況說明，并提供身份證明、權屬證明及詳細侵權或不實情況證明。本網(wǎng)站在收到上述法律文件后，將會依法盡快聯(lián)系相關文章源頭核實，溝通刪除相關內(nèi)容或斷開相關鏈接。