PySpark

1. Create a main program file.

In this topic, you create a file named example.py. This file contains the following code. The main function in the sample code allows PySpark to find the entry point to start a program.

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. Execute the main program file.

1. Prepare test data.

Create a CSV file named staff.csv, and upload the file to OSS. This file contains the following content that indicates the basic information and income of each employee.

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

2. Develop a dependency method.

1. Create a folder named tools.

2. Create a file named func.py in the tools folder. This file contains the following content.

   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. Package the tools folder into the tools.zip file, and upload the file to OSS. The following code is used to generate the tools.zip file.

   Notice

   The ZIP compression tools vary based on the operating system that you use. Make sure that the tools folder is the root directory after the tools.zip file is decompressed.

3. Develop the main program.

Develop a Spark program in Python. This program allows DLA to read data from the staff.csv file in OSS. Then, register this file as a DataFrame. In addition, register the tax method in the package that is created in Step 2 as a Spark UDF.Then, use the Spark UDF to calculate the DataFrame and generate results.

In the following sample code, use the name of your OSS bucket to replace {your bucket name}.

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. Submit a job.

In the DLA console, choose Serverless Spark > Submit job from the left-side navigation pane. Then, create a job and enter the following code in the code editor:

{
    "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"
    }
}

The following table describes the parameters in the code.

Parameter	Description	Required
conf	The configuration parameters that are required for the Spark job. "spark.dla.connectors": "oss": indicates that the OSS connector is used. "spark.kubernetes.pyspark.pythonVersion": "3": indicates that Python 3 is used to run the job.	No

1. Prepare a Linux operating system.

The operating system for packaging a virtual environment must be the same as the operating system of the serverless Spark engine where the virtual environment package is used. Therefore, the virtual environment package that needs to be uploaded to the serverless Spark engine of DLA must be installed in Linux. You can use one of the following methods to prepare a Linux operating system for packaging a virtual environment:

2. Package a Python virtual environment in Linux.

You can use a tool, such as virtualenv or conda, to package a virtual environment. Before you package a virtual environment, select a tool based on your business requirements and install the tool in Linux.

The following code demonstrates how to use virtualenv to generate a venv.zip file. The venv.zip file contains scikit-spark of a specific version.

# 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. Run the virtual environment in the serverless Spark engine.

Before you submit a Spark job, use the following code to configure the job: In the code, the spark.pyspark.python parameter specifies the file that is used to run the virtual environment in the uploaded package. For more information about the parameters, see Configure a Spark job

{
    "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"
}