AnalyticDB:Use Airflow to schedule Spark jobs

Batch mode

Interactive mode

1. Obtain the connection URL of the Spark interactive resource group.

   1. Log on to the AnalyticDB for MySQL console. In the upper-left corner of the console, select a region. In the left-side navigation pane, click Clusters. On the Enterprise Edition, Basic Edition, or Data Lakehouse Edition tab, find the cluster that you want to manage and click the cluster ID.

   2. In the left-side navigation pane, choose Cluster Management > Resource Management. On the page that appears, click the Resource Groups tab.

   3. Find the Spark interactive resource group that you created and click Details in the Actions column to view the internal or public connection URL of the resource group. You can click the icon within the parentheses next to the corresponding port number to copy the connection URL.

      You must click Apply for Endpoint next to Public Endpoint to manually apply for a public endpoint in the following scenarios:

      • The client tool that is used to submit a Spark SQL job is deployed on an on-premises or external server.

      • The client tool that is used to submit a Spark SQL job is deployed on an Elastic Compute Service (ECS) instance that resides in a different virtual private cloud (VPC) from your AnalyticDB for MySQL cluster.

2. Install the apache-airflow-providers-apache-hive and apache-airflow-providers-common-sql dependencies.

3. Access the Airflow web interface. In the top navigation bar, choose Admin > Connections.

4. Click the button. On the Add Connections page, configure the parameters that are described in the following table.

   Parameter	Description
   Connection Id	The name of the connection. In this example, adb_spark_cluster is used.
   Connection Type	The type of the connection. Select Hive Server 2 Thrift.
   Host	The endpoint obtained in Step 1. Replace default in the endpoint with the actual name of the database and delete the resource_group=<resource group name> suffix from the endpoint. Example: jdbc:hive2://amv-t4naxpqk****sparkwho.ads.aliyuncs.com:10000/adb_demo.
   Schema	The name of the database. In this example, adb_demo is used.
   Login	The names of the database account and the resource group in the AnalyticDB for MySQL cluster. Format: resource_group_name/database_account_name. Example: spark_interactive_prod/spark_user.
   Password	The password of the database account of the AnalyticDB for MySQL cluster.
   Port	The port number for Spark interactive resource groups. Set the value to 10000.
   Extra	The authentication method. Enter the following content, which specifies that username and password authentication is used: { "auth_mechanism": "CUSTOM" }

5. Write a DAG file.

   from airflow import DAG
   from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
   from datetime import datetime
   
   default_args = {
       'owner': 'airflow',
       'start_date': datetime(2025, 2, 10),
       'retries': 1,
   }
   
   dag = DAG(
       'adb_spark_sql_test',
       default_args=default_args,
       schedule_interval='@daily',
   )
   
   
   jdbc_query = SQLExecuteQueryOperator(
       task_id='execute_spark_sql_query', 
       conn_id='adb_spark_cluster',  
       sql='show databases',  
       dag=dag
   )
   
   jdbc_query

   The following table describes the parameters.

   Parameter	Required	Description
   task_id	Yes	The job ID. You can enter a custom ID.
   conn_id	Yes	The name of the connection. Enter the ID of the connection that you created in Step 4.
   sql	Yes	The Spark SQL statement.

   For more information about optional parameters, see Airflow parameters.

6. On the Airflow web interface, click the button next to the DAG.