Start a Spark task-E-MapReduce(EMR)-阿里云帮助中心

Submit Spark tasks to E-MapReduce (EMR) Serverless Spark programmatically using the Python SDK.

Prerequisites

Before you begin, ensure that you have:

An AccessKey pair. Use a RAM user's AccessKey pair scoped to EMR Serverless Spark rather than your Alibaba Cloud account AccessKey pair. To set this up:
- Create a RAM user, then create an AccessKey pair for the RAM user
- Grant EMR Serverless Spark permissions to the RAM user
Python 3
The ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variables set to your AccessKey pair. See Configure environment variables on Linux, macOS, and Windows

Using environment variables for credentials keeps your AccessKey pair out of source code. For production workloads, use Security Token Service (STS) temporary credentials for better security.

Step 1: Install the SDK

pip install alibabacloud_emr_serverless_spark20230808==1.0.0

Step 2: Configure the client

All examples use the following client setup. Replace <region-id> with the region where your EMR Serverless Spark workspace is deployed (for example, cn-hangzhou). For a full list of endpoints, see Endpoints.

# -*- coding: utf-8 -*-
import os
from typing import List
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_emr_serverless_spark20230808.client import Client
from alibabacloud_emr_serverless_spark20230808.models import (
    StartJobRunRequest,
    Tag,
    JobDriver,
    JobDriverSparkSubmit,
)
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


def create_client() -> Client:
    config = open_api_models.Config(
        access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
        access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
    )
    config.endpoint = 'emr-serverless-spark.<region-id>.aliyuncs.com'
    return Client(config)

Step 3: Submit a job run

Call start_job_run_with_options to submit a Spark job run to your workspace.

Required parameters

Parameter	Description
`region_id`	The region ID of your workspace, for example, `cn-hangzhou`
`resource_queue_id`	The resource queue to run the job in, for example, `root_queue`
`code_type`	The type of Spark job: `JAR`, `PYTHON`, or `SQL`
`name`	A name for the job run
`release_version`	The Spark runtime version, for example, `esr-2.1-native (Spark 3.3.1, Scala 2.12, Native Runtime)`
`job_driver`	A `JobDriver` wrapping a `JobDriverSparkSubmit` that defines the entry point and Spark configuration

JobDriverSparkSubmit parameters

JobDriverSparkSubmit accepts three positional arguments:

Argument	Description
`entry_point`	The OSS path to the JAR file, Python script, or SQL file to run
`entry_point_arguments`	A list of arguments passed to the entry point
`spark_submit_parameters`	Additional Spark parameters such as `--class`, `--conf spark.executor.cores`, and `--conf spark.executor.memory`

For JAR jobs, specify the entry class in spark_submit_parameters using --class.

Example: JAR job

Replace the following placeholders before running:

Placeholder	Description	Example
`<workspace-id>`	Your EMR Serverless Spark workspace ID	`w-ae42e9c92927****`
`<region-id>`	Region ID	`cn-hangzhou`
`<YourBucket>`	Your OSS bucket name	`my-emr-bucket`

def run_jar_job():
    client = create_client()
    tags: List[Tag] = [Tag("environment", "production"), Tag("workflow", "true")]
    job_driver_spark_submit = JobDriverSparkSubmit(
        "oss://<YourBucket>/spark-resource/examples/jars/spark-examples_2.12-3.3.1.jar",
        ["1"],
        "--class org.apache.spark.examples.SparkPi --conf spark.executor.cores=4 --conf spark.executor.memory=20g --conf spark.driver.cores=4 --conf spark.driver.memory=8g --conf spark.executor.instances=1"
    )
    job_driver = JobDriver(job_driver_spark_submit)
    request = StartJobRunRequest(
        region_id="<region-id>",
        resource_queue_id="root_queue",
        code_type="JAR",
        name="emr-spark-task",
        release_version="esr-2.1-native (Spark 3.3.1, Scala 2.12, Native Runtime)",
        tags=tags,
        job_driver=job_driver
    )
    runtime = util_models.RuntimeOptions()
    headers = {}
    try:
        response = client.start_job_run_with_options('<workspace-id>', request, headers, runtime)
        print(response.body.to_map())
    except Exception as error:
        print(error.message)
        print(error.data.get("Recommend"))
        UtilClient.assert_as_string(error.message)


run_jar_job()

Example: Python job

Use the same placeholders as the JAR example (<workspace-id>, <region-id>, <YourBucket>).

def run_python_job():
    client = create_client()
    tags: List[Tag] = [Tag("environment", "production"), Tag("workflow", "true")]
    job_driver_spark_submit = JobDriverSparkSubmit(
        "oss://<YourBucket>/spark-resource/examples/src/main/python/pi.py",
        ["50"],
        "--conf spark.executor.cores=4 --conf spark.executor.memory=20g --conf spark.driver.cores=4 --conf spark.driver.memory=8g --conf spark.executor.instances=1"
    )
    job_driver = JobDriver(job_driver_spark_submit)
    request = StartJobRunRequest(
        region_id="<region-id>",
        resource_queue_id="root_queue",
        code_type="PYTHON",
        name="emr-spark-task",
        release_version="esr-2.1-native (Spark 3.3.1, Scala 2.12, Native Runtime)",
        tags=tags,
        job_driver=job_driver
    )
    runtime = util_models.RuntimeOptions()
    headers = {}
    try:
        response = client.start_job_run_with_options('<workspace-id>', request, headers, runtime)
        print(response.body.to_map())
    except Exception as error:
        print(error.message)
        print(error.data.get("Recommend"))
        UtilClient.assert_as_string(error.message)


run_python_job()

Example: SQL job

Use the same placeholders as the JAR example. For SQL jobs, pass the OSS path to your SQL file as both the entry_point and the first element of entry_point_arguments (using the -f flag).

def run_sql_job():
    client = create_client()
    tags: List[Tag] = [Tag("environment", "production"), Tag("workflow", "true")]
    job_driver_spark_submit = JobDriverSparkSubmit(
        "oss://<YourBucket>/spark-resource/examples/sql/show_db.sql",
        ["-f", "oss://<YourBucket>/spark-resource/examples/sql/show_db.sql"],
        "--class org.apache.spark.sql.hive.thriftserver.SparkSQLCLIDriver --conf spark.executor.cores=4 --conf spark.executor.memory=20g --conf spark.driver.cores=4 --conf spark.driver.memory=8g --conf spark.executor.instances=1"
    )
    job_driver = JobDriver(job_driver_spark_submit)
    request = StartJobRunRequest(
        region_id="<region-id>",
        resource_queue_id="root_queue",
        code_type="SQL",
        name="airflow-sql-test",
        release_version="esr-2.1-native (Spark 3.3.1, Scala 2.12, Native Runtime)",
        tags=tags,
        job_driver=job_driver
    )
    runtime = util_models.RuntimeOptions()
    headers = {}
    try:
        response = client.start_job_run_with_options('<workspace-id>', request, headers, runtime)
        print(response.body.to_map())
    except Exception as error:
        print(error.message)
        print(error.data.get("Recommend"))
        UtilClient.assert_as_string(error.message)


run_sql_job()