E-MapReduce:Use Apache Airflow to submit a job

Step 1: Create a Livy gateway and a token

1. Create and start a Livy gateway.

   1. Go to the Gateways page.

      1. Log on to the EMR console.

      2. In the left-side navigation pane, choose EMR Serverless > Spark.

      3. On the Spark page, find the desired workspace and click the name of the workspace.

      4. In the left-side navigation pane of the EMR Serverless Spark page, choose Operation Center > Gateways.

   2. On the Gateways page, click the Livy Gateways tab.

   3. On the Livy Gateways tab, click Create Livy Gateway.

   4. On the Create Livy Gateway page, configure the Name parameter and click Create. In this example, the gateway name is Livy-gateway.

      You can configure other parameters based on your business requirements. For more information, see Manage gateways.

   5. On the Livy Gateways tab, find the created Livy gateway and click Start in the Actions column.

2. Create a token for the gateway.

   1. On the Livy Gateways tab, find the created Livy gateway and click Tokens in the Actions column.

   2. On the Tokens tab, click Create Token.

   3. In the Create Token dialog box, configure the Name parameter and click OK. In this example, the token name is Livy-token.

   4. Copy the token.

      Important

      After you create the token, you must immediately copy the token. You can no longer view the token after you leave the page. If the token expires or is lost, reset the token or create a new token.

Step 2: Configure Apache Airflow

1. Run the following command to install Apache Livy in Apache Airflow:

   pip install apache-airflow-providers-apache-livy

2. Add a connection.

   Use the UI

   Find the default connection named livy_default in Apache Airflow and modify the connection. Alternatively, you can manually add a connection on the web UI of Apache Airflow. For more information, see Creating a Connection with the UI.

   You must configure the following parameters:

   • Host: Enter the endpoint of the gateway.

   • Schema: Enter https.

   • Extra: Enter a JSON string. x-acs-spark-livy-token indicates the token that you copied in Step 1.

     {
       "x-acs-spark-livy-token": "6ac**********kfu"
     }

   Use the CLI

   Use the Airflow command-line interface (CLI) to run a command to add a connection. For more information, see Creating a Connection.

   airflow connections add 'livy_default' \
       --conn-json '{
           "conn_type": "livy",
           "host": "pre-emr-spark-livy-gateway-cn-hangzhou.data.aliyun.com/api/v1/workspace/w-xxxxxxx/livycompute/lc-xxxxxxx",   # The endpoint of the gateway.
           "schema": "https",
           "extra": {
               "x-acs-spark-livy-token": "6ac**********kfu"  # The token that you copied in Step 1.
           }
       }'

Step 3: Configure a DAG and submit a job

Apache Airflow allows you to configure a DAG to declare how a job runs. The following sample code provides an example on how to use Livy Operator to run a Spark job.

Obtain the Python file from Object Storage Service (OSS) and execute the file.

from datetime import timedelta, datetime
from airflow import DAG
from airflow.providers.apache.livy.operators.livy import LivyOperator

default_args = {
    'owner': 'aliyun',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# Initiate DAG
livy_operator_sparkpi_dag = DAG(
    dag_id="livy_operator_sparkpi_dag",  # The unique ID of the DAG.
    default_args=default_args,
    schedule_interval=None,
    start_date=datetime(2024, 5, 20),
    tags=['example', 'spark', 'livy'],
    catchup=False
)

# define livy task with LivyOperator
# Replace the content of the file based on your business requirements.
livy_sparkpi_submit_task = LivyOperator(
    file="oss://<YourBucket>/jars/spark-examples_2.12-3.3.1.jar",
    class_name="org.apache.spark.examples.SparkPi",
    args=['1000'],
    driver_memory="1g",
    driver_cores=1,
    executor_memory="1g",
    executor_cores=2,
    num_executors=1,
    name="LivyOperator SparkPi",
    task_id="livy_sparkpi_submit_task", 
    dag=livy_operator_sparkpi_dag,
)

livy_sparkpi_submit_task

The following table describes the parameters in the preceding code.