通過PySpark開發Spark應用 - AnalyticDB

本文介紹了如何開發 AnalyticDB for MySQL Spark Python作業，以及如何通過VirtualEnv技術打包Python作業的運行環境。

前提條件

叢集的產品系列為企業版、基礎版或湖倉版。
叢集與OSS儲存空間位於相同地區。
已在企業版、基礎版或湖倉版叢集中建立Job型資源群組。
已建立 AnalyticDB for MySQL 叢集的資料庫帳號。
- 如果是通過阿里雲帳號訪問，只需建立高許可權帳號。
- 如果是通過RAM使用者訪問，需要建立高許可權帳號和普通帳號並且將 RAM使用者綁定到普通帳號上。

PySpark的基本用法

編寫如下樣本程式，並將樣本程式儲存為 example.py 。

from pyspark.sql import SparkSession
if __name__ == "__main__":
    spark = SparkSession.builder.getOrCreate()
    df = spark.sql("SELECT 1+1")
    df.printSchema()
    df.show()

將 example.py 程式上傳到OSS中。具體操作，請參見控制台上傳檔案。
進入Spark開發編輯器。
1. 登入雲原生資料倉儲AnalyticDB MySQL控制台，在左上方選擇叢集所在地區。在左側導覽列，單擊 集群清單 ，然後單擊目的地組群ID。
2. 在左側導覽列，單擊 作業開發 > Spark Jar 開發 。
在編輯器視窗上方，選擇Job型資源群組和Spark作業類型。本文以Batch類型為例。

在編輯器中執行以下作業內容。

{
 "name": "Spark Python Test",
 "file": "oss://testBucketName/example.py",
 "conf": {
 "spark.driver.resourceSpec": "small",
 "spark.executor.instances": 1,
 "spark.executor.resourceSpec": "small"
}

參數說明請參見參數說明。

使用Python依賴

使用方法

如果您使用自行開發或第三方開發的依賴開發Python程式時，需將使用的依賴上傳至OSS中，並在提交Spark作業時配置 pyFiles 參數。

樣本

本文樣本以引入自訂Function Compute員工的稅後收入為例。樣本將資料檔案 staff.csv 上傳至OSS中。 staff.csv 中的樣本資料如下：

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

開發依賴並上傳至OSS中。
1. 建立名為 tools 的檔案夾，並在該檔案夾下建立名為 func.py 的程式。
```
def tax(salary):
    convert string to int and cut 15% tax from the salary
    :param salary: The salary of staff worker
    :return:
    return 0.15 * int(salary)
```
2. 將 tools 檔案夾壓縮後上傳至 OSS中。本文樣本為 tools.zip 。

from __future__ import print_function
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
from pyspark.sql.types import FloatType
import sys
# import third party 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
    cvs_file = sys.argv[1]
    df = spark.read.csv(cvs_file, 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()

```
{
 "name": "Spark Python",
 "file": "oss://testBucketName/example.py",
 "pyFiles": ["oss://testBucketName/tools.zip"],
 "args": [
 "oss://testBucketName/staff.csv"
 "conf": {
 "spark.driver.resourceSpec": "small",
 "spark.executor.instances": 2,
 "spark.executor.resourceSpec": "small"
}
```

# 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 party modules
pip install scikit-spark==0.4.0
# check the result
pip list
# compress the environment
tar -czvf venv.tar.gz venv

{
 "name": "venv example",
 "archives": [
 "oss://testBucketname/venv.tar.gz#PY3"
 "conf": {
 "spark.driver.resourceSpec": "small",
 "spark.executor.instances": 1,
 "spark.pyspark.python": "./PY3/venv/bin/python3",
 "spark.executor.resourceSpec": "small"
 "file": "oss://testBucketname/example.py"
}

{
 "name": "venv example",
 "conf": {
 "spark.executorEnv.ADB_SPARK_DOWNLOAD_FILES": "oss://testBucketname/venv_py36.tar.gz#PY3",
 "spark.kubernetes.driverEnv.ADB_SPARK_DOWNLOAD_FILES": "oss://atestBucketname/venv_py36.tar.gz#PY3,",
 "spark.driver.resourceSpec": "small",
 "spark.executor.instances": 1,
 "spark.pyspark.python": "./PY3/venv/bin/python3",
 "spark.executor.resourceSpec": "small"
 "file": "oss://testBucketname/example.py"
}

前提條件

PySpark的基本用法

使用Python依賴

使用方法

樣本

使用Virtual Environments打包依賴環境

使用方法

樣本