DMSAnalyticDBSparkOperator - Data Management - Alibaba Cloud Documentation Center

This topic describes the parameters for the DMSAnalyticDBSparkOperator.

Overview

The DMSAnalyticDBSparkOperator is a general-purpose operator for AnalyticDB Spark. It supports both Warehouse and Job resource group types, as well as SQL, STREAMING, and BATCH application types. For more information, see Application scenarios.

Parameters

Note

The parameters sql, conf, cluster_id, resource_group, instance, schema, app_name, app_type, and execute_time_limit_in_seconds can use Jinja templates.

Parameter	Type	Required	Description
cluster_id	string	Yes	`cluster_id`: The ID of the AnalyticDB for MySQL cluster. `instance`: The database connection (DBLink) managed by DMS. Note You must specify either `cluster_id` or `instance`. We recommend that you use `instance`.
instance	string	Yes
resource_group	string	Yes	The name of the resource group for the AnalyticDB for MySQL cluster.
region	string	No	The region ID. This parameter is required for cross-region calls.
app_name	string	No	The application name. This parameter is used when the resource group type is Job.
app_type	string	No	The application type. This parameter is used when the resource group type is Job. Valid values: `SQL`: Runs an SQL query. `STREAMING`: A streaming task. `BATCH`: A batch task.
sql	string	Yes	The Spark SQL statement to execute.
conf	dict	No	Custom Spark configurations. For more information, see ExecuteSparkWarehouseBatchSQL, GetSparkWarehouseBatchSQL, and CancelSparkWarehouseBatchSQL. Note The `conf` parameter is the ExecuteSparkWarehouseBatchSQL `RuntimeConfig`.
schema	string	No	The database to use. The default is `default`.
polling_interval	int	No	The polling interval, in seconds, for the execution status. The default is 10. If set to 0 or a negative number, the operator submits the task and exits without waiting for completion. Status polling includes a built-in retry mechanism.
execute_time_limit_in_seconds	int	No	The execution timeout in seconds. The default is 36000 (10 hours).
callback	function	No	A callback function that processes the SQL result. The input parameter is SparkBatchSQL. Note This parameter takes effect only when `polling_interval` is greater than 0.

Example

Note

task_id and dag are Airflow-specific parameters. For more information, see the official Airflow documentation.

from typing import Any
from airflow import DAG
from airflow.decorators import task
from airflow.models.param import Param
from airflow.operators.empty import EmptyOperator
from datetime import datetime

from airflow.providers.alibabadms.cloud.operators.dms_analyticdb_spark import DMSAnalyticDBSparkOperator

# alibabacloud_adb20211201.models.SparkBatchSQL
def print_result(result):
    print(f"{result}")

with DAG(
    "dms_adb_spark_dblink",
    params={
        "sql": "show databases;show databases;"
    }
) as dag:

    warehouse_operator: Any = DMSAnalyticDBSparkOperator(
        task_id="warehouse_sql",
        instance="dbl_adbmysql_89",
        resource_group="hansheng_spark_test",
        sql="{{ params.sql }}",
        polling_interval=5,
        conf={
            'spark.adb.sqlOutputFormat':'CSV',
            'spark.adb.sqlOutputPartitions':1,
            'spark.adb.sqlOutputLocation':'oss://hansheng-bj/airflow/adb_spark/test',
            'sep':'|'
        },
        callback=print_result,
        dag=dag
    )

    run_this_last = EmptyOperator(
        task_id="run_this_last",
        dag=dag
    )

    warehouse_operator >> run_this_last

if __name__ == "__main__":
    dag.test(
        run_conf={}
    )

Note

All DMS Airflow operators support common features such as task cancellation and automatic retries. For more information, see Airflow DMS Operator.