Data Lake Analytics - Deprecated：PySpark

1.開發主程式檔案

您可以建立如下內容的example.py檔案，樣本中定義main函數可以允許PySpark找到程式的統一啟動入口。

from __future__ import print_function
from pyspark.sql import SparkSession


# import third part file
from tools import func

if __name__ == "__main__":
    # init pyspark context
    spark = SparkSession\
        .builder\
        .appName("Python Example")\
        .getOrCreate()

    df = spark.sql("SELECT 2021")
    # print schema and data to the console
    df.printSchema()
    df.show()

2.執行主程式檔案

1. 和Scala、Java程式開發的JAR包一樣，您需要將example.py檔案上傳到OSS中，並在Spark的啟動配置中使用file來指定這個檔案為開機檔案。

2. 在DLA控制台的Serverless->作業管理頁面，使用如下範例程式碼配置作業。

   {
   
       "name": "Spark Python",
   
       "file": "oss://{your bucket name}/example.py"
   
       "conf": {
   
           "spark.driver.resourceSpec": "small",
   
           "spark.executor.instances": 2,
   
           "spark.executor.resourceSpec": "small",
   
           "spark.kubernetes.pyspark.pythonVersion": "3"
   
     }
   }

   重要

   • 您需要在配置時將{your bucket name}替換為您使用的OSS的Bucket名稱。

   • 樣本中使用Python 3執行檔案，和社區版的Spark相同，通過spark.kubernetes.pyspark.pythonVersion配置使用的Python版本，預設為Python 2.7。

3. 單擊執行。

1. 準備測試資料

建立一個如下格式的CSV檔案，命名為staff.csv，並上傳到OSS中。檔案反映了每個員工的資訊和收入情況。

name,age,gender,salary
Lucky,25,male,100
Lucy,23,female,150
Martin,30,male,180
Rose,31,female,200

2. 開發一個依賴方法

1. 建立一個檔案夾tools。

2. 在tools檔案夾中建立一個檔案func.py，檔案內容如下。

   def tax(salary):
       """
       convert string to int
       then cut 15% tax from the salary
       return a float number
   
       :param salary: The salary of staff worker
       :return:
       """
       return 0.15 * int(salary)

3. 將tools檔案夾壓縮為tools.zip後上傳到OSS中。壓縮包的產生方式如下。

   重要

   不同作業系統平台的ZIP壓縮公用程式會略有區別，請保證解壓後可以看到頂層目錄是tools檔案夾。

3. 開發主程式

開發一個Spark的Python程式，將測試中的CSV從OSS中讀取出來，註冊為一個DataFrame。同時將依賴包中的tax方法註冊為一個Spark UDF，然後使用該UDF對剛剛產生的DataFrame進行計算並列印結果。

範例程式碼如下， 您需要在配置時將{your bucket name}替換為您使用的OSS的Bucket名稱。

from __future__ import print_function
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import FloatType

# import third part file
from tools import func

if __name__ == "__main__":
    # init pyspark context
    spark = SparkSession\
        .builder\
        .appName("Python Example")\
        .getOrCreate()

    # read csv from oss to a dataframe, show the table
    df = spark.read.csv('oss://{your bucket}/staff.csv', mode="DROPMALFORMED",inferSchema=True, header = True)
    # print schema and data to the console
    df.printSchema()
    df.show()

    # create an udf
    taxCut = udf(lambda salary: func.tax(salary), FloatType())

    # cut tax from salary and show result
    df.select("name", taxCut("salary").alias("final salary")).show()
    spark.stop()

4. 提交任務

在DLA控制台的Serverless->作業管理頁面，建立一個作業，並提交以下作業資訊。

{
    "name": "Spark Python",
    "file": "oss://{your bucket name}/example.py",
    "pyFiles": ["oss://{your bucket name}/tools.zip"],
    "conf": {
        "spark.driver.resourceSpec": "small",
        "spark.executor.instances": 2,
        "spark.executor.resourceSpec": "small",
        "spark.dla.connectors": "oss",
        "spark.kubernetes.pyspark.pythonVersion": "3"
    }
}

代碼中主要參數說明。

1.準備Linux環境

由於Virtualenv需要相同的作業系統，當前上傳到DLA Spark使用的壓縮包必須在Linux環境下的進行安裝。您可以採用如下方式準備Linux環境。

• 準備一台Centos7的電腦進行打包。

• 在阿里雲以隨用隨付的方式新開一台Centos 7的ECS，使用完畢後關閉。

• 使用Centos 7的官方Docker鏡像，在鏡像內部打包。

2.在Linux環境下打包Python執行環境

常用的執行環境打包工具包括Virtualenv、Conda，您可以根據您的需要來選擇對應的工具，並安裝工具到您的Linux環境中。

以下樣本使用Virtualenv產生一個執行環境壓縮包venv.zip，壓縮包中包含了scikit-spark的特定版本。

# create directory venv at current path with python3
# MUST ADD --copies !
virtualenv --copies --download --python Python3.7 venv

# active environment
source venv/bin/activate

# install third part modules
pip install scikit-spark==0.4.0

# check the result
pip list

# zip the environment
zip -r venv.zip venv

3.在Spark中使用Python執行環境

您可以在提交Spark作業時，使用如下的代碼配置作業。其中spark.pyspark.python的參數值表示上傳的壓縮檔中的運行包。更多參數說明，請參見作業參數說明。

{
    "name": "venv example",
    "archives": [
        "oss://test/venv.zip#PY3"
    ],
    "conf": {
        "spark.driver.resourceSpec": "medium",
        "spark.dla.connectors": "oss",
        "spark.executor.instances": 1,
        "spark.dla.job.log.oss.uri": "oss://test/spark-logs",
        "spark.pyspark.python": "./PY3/venv/bin/python3",
        "spark.executor.resourceSpec": "medium"
    },
    "file": "oss://test/example.py"
}