Python Spark的實現(xiàn)原理是什么

Python Spark的實現(xiàn)原理是什么，相信很多沒有經(jīng)驗的人對此束手無策，為此本文總結(jié)了問題出現(xiàn)的原因和解決方法，通過這篇文章希望你能解決這個問題。

Spark運行時架構(gòu)

首先我們先回顧下Spark的基本運行時架構(gòu)，如下圖所示，其中橙色部分表示為JVM，Spark應用程序運行時主要分為Driver和Executor，Driver負載總體調(diào)度及UI展示，Executor負責Task運行，Spark可以部署在多種資源管理系統(tǒng)中，例如Yarn、Mesos等，同時Spark自身也實現(xiàn)了一種簡單的Standalone(獨立部署)資源管理系統(tǒng)，可以不用借助其他資源管理系統(tǒng)即可運行。

用戶的Spark應用程序運行在Driver上(某種程度上說，用戶的程序就是Spark Driver程序)，經(jīng)過Spark調(diào)度封裝成一個個Task，再將這些Task信息發(fā)給Executor執(zhí)行，Task信息包括代碼邏輯以及數(shù)據(jù)信息，Executor不直接運行用戶的代碼。

PySpark運行時架構(gòu)

為了不破壞Spark已有的運行時架構(gòu)，Spark在外圍包裝一層Python API，借助Py4j實現(xiàn)Python和Java的交互，進而實現(xiàn)通過Python編寫Spark應用程序，其運行時架構(gòu)如下圖所示。

其中白色部分是新增的Python進程，在Driver端，通過Py4j實現(xiàn)在Python中調(diào)用Java的方法，即將用戶寫的PySpark程序”映射”到JVM中，例如，用戶在PySpark中實例化一個Python的SparkContext對象，最終會在JVM中實例化Scala的SparkContext對象；在Executor端，則不需要借助Py4j，因為Executor端運行的Task邏輯是由Driver發(fā)過來的，那是序列化后的字節(jié)碼，雖然里面可能包含有用戶定義的Python函數(shù)或Lambda表達式，Py4j并不能實現(xiàn)在Java里調(diào)用Python的方法，為了能在Executor端運行用戶定義的Python函數(shù)或Lambda表達式，則需要為每個Task單獨啟一個Python進程，通過socket通信方式將Python函數(shù)或Lambda表達式發(fā)給Python進程執(zhí)行。語言層面的交互總體流程如下圖所示，實線表示方法調(diào)用，虛線表示結(jié)果返回。

下面分別詳細剖析PySpark的Driver是如何運行起來的以及Executor是如何運行Task的。

Driver端運行原理

當我們通過spark-submmit提交pyspark程序，首先會上傳python腳本及依賴，并申請Driver資源，當申請到Driver資源后，會通過PythonRunner(其中有main方法)拉起JVM，如下圖所示。

PythonRunner入口main函數(shù)里主要做兩件事：

• 開啟Py4j GatewayServer

• 通過Java Process方式運行用戶上傳的Python腳本

用戶Python腳本起來后，首先會實例化Python版的SparkContext對象，在實例化過程中會做兩件事：

• 實例化Py4j GatewayClient，連接JVM中的Py4j GatewayServer，后續(xù)在Python中調(diào)用Java的方法都是借助這個Py4j Gateway

• 通過Py4j Gateway在JVM中實例化SparkContext對象

經(jīng)過上面兩步后，SparkContext對象初始化完畢，Driver已經(jīng)起來了，開始申請Executor資源，同時開始調(diào)度任務。用戶Python腳本中定義的一系列處理邏輯最終遇到action方法后會觸發(fā)Job的提交，提交Job時是直接通過Py4j調(diào)用Java的PythonRDD.runJob方法完成，映射到JVM中，會轉(zhuǎn)給sparkContext.runJob方法，Job運行完成后，JVM中會開啟一個本地Socket等待Python進程拉取，對應地，Python進程在調(diào)用PythonRDD.runJob后就會通過Socket去拉取結(jié)果。

把前面運行時架構(gòu)圖中Driver部分單獨拉出來，如下圖所示，通過PythonRunner入口main函數(shù)拉起JVM和Python進程，JVM進程對應下圖橙色部分，Python進程對應下圖白色部分。Python進程通過Py4j調(diào)用Java方法提交Job，Job運行結(jié)果通過本地Socket被拉取到Python進程。還有一點是，對于大數(shù)據(jù)量，例如廣播變量等，Python進程和JVM進程是通過本地文件系統(tǒng)來交互，以減少進程間的數(shù)據(jù)傳輸。

Executor端運行原理

為了方便闡述，以Spark On Yarn為例，當Driver申請到Executor資源時，會通過CoarseGrainedExecutorBackend(其中有main方法)拉起JVM，啟動一些必要的服務后等待Driver的Task下發(fā)，在還沒有Task下發(fā)過來時，Executor端是沒有Python進程的。當收到Driver下發(fā)過來的Task后，Executor的內(nèi)部運行過程如下圖所示。

Executor端收到Task后，會通過launchTask運行Task，最后會調(diào)用到PythonRDD的compute方法，來處理一個分區(qū)的數(shù)據(jù)，PythonRDD的compute方法的計算流程大致分三步走：

• 如果不存在pyspark.deamon后臺Python進程，那么通過Java Process的方式啟動pyspark.deamon后臺進程，注意每個Executor上只會有一個pyspark.deamon后臺進程，否則，直接通過Socket連接pyspark.deamon，請求開啟一個pyspark.worker進程運行用戶定義的Python函數(shù)或Lambda表達式。pyspark.deamon是一個典型的多進程服務器，來一個Socket請求，fork一個pyspark.worker進程處理，一個Executor上同時運行多少個Task，就會有多少個對應的pyspark.worker進程。

• 緊接著會單獨開一個線程，給pyspark.worker進程喂數(shù)據(jù)，pyspark.worker則會調(diào)用用戶定義的Python函數(shù)或Lambda表達式處理計算。

• 在一邊喂數(shù)據(jù)的過程中，另一邊則通過Socket去拉取pyspark.worker的計算結(jié)果。

把前面運行時架構(gòu)圖中Executor部分單獨拉出來，如下圖所示，橙色部分為JVM進程，白色部分為Python進程，每個Executor上有一個公共的pyspark.deamon進程，負責接收Task請求，并fork pyspark.worker進程單獨處理每個Task，實際數(shù)據(jù)處理過程中，pyspark.worker進程和JVM Task會較頻繁地進行本地Socket數(shù)據(jù)通信。

總體上來說，PySpark是借助Py4j實現(xiàn)Python調(diào)用Java，來驅(qū)動Spark應用程序，本質(zhì)上主要還是JVM runtime，Java到Python的結(jié)果返回是通過本地Socket完成。雖然這種架構(gòu)保證了Spark核心代碼的獨立性，但是在大數(shù)據(jù)場景下，JVM和Python進程間頻繁的數(shù)據(jù)通信導致其性能損耗較多，惡劣時還可能會直接卡死，所以建議對于大規(guī)模機器學習或者Streaming應用場景還是慎用PySpark，盡量使用原生的Scala/Java編寫應用程序，對于中小規(guī)模數(shù)據(jù)量下的簡單離線任務，可以使用PySpark快速部署提交。

看完上述內(nèi)容，你們掌握Python Spark的實現(xiàn)原理是什么的方法了嗎？如果還想學到更多技能或想了解更多相關(guān)內(nèi)容，歡迎關(guān)注億速云行業(yè)資訊頻道，感謝各位的閱讀！