DMSAnalyticDBSparkSqlOperator的参数与示例-数据管理 DMS-阿里云-数据管理 DMS-阿里云帮助中心

本文为您介绍DMSAnalyticDBSparkOperator操作的配置信息。

功能说明

AnalyticDB Spark通用Operator，同时支持Warehouse和Job类型资源组，支持SQL、STREAMING、BATCH应用类型。详情请参见应用场景。

参数说明

说明

参数sql、conf、cluster_id、resource_group、instance、schema、app_name、app_type、execute_time_limit_in_seconds可以使用Jinja模板。

参数	类型	是否必填	说明
cluster_id	string	是	cluster_id：AnalyticDB MySQL集群的ID。 instance：DMS管理的数据库实例的连接（DBLink），详情请参见逻辑数仓。说明二选一，建议使用`instance`。
instance	string	是
resource_group	string	是	AnalyticDB MySQL集群的资源组名称。
region	string	否	Region ID，跨地域调用时需指定。
app_name	string	否	应用名称，Job类型资源组时使用。
app_type	string	否	应用类型，Job类型资源组时使用。取值： `SQL`：执行SQL。 `STREAMING`：流式任务。 `BATCH`：批量任务。
sql	string	是	Spark的SQL操作。
conf	dict	否	Spark的特殊配置，详情请参见ExecuteSparkWarehouseBatchSQL、GetSparkWarehouseBatchSQL、CancelSparkWarehouseBatchSQL。说明 `conf`参数即为ExecuteSparkWarehouseBatchSQL中的`RuntimeConfig`。
schema	string	否	数据库，默认为default。
polling_interval	int	否	刷新执行结果的间隔时间。单位为秒，默认值为10。设置为0或负数时，任务仅提交不等待结果。状态查询内置重试机制。
execute_time_limit_in_seconds	int	否	超时时间，单位为秒，默认值为36000（10小时）。
callback	function	否	用于处理SQL操作结果的回调函数，入参为SparkBatchSQL。说明当`polling_interval`的值大于0时，本参数才生效。

示例

说明

task_id和dag是Airflow的特定参数，详情请参见Airflow官方文档。

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 = DMSAnalyticDBSparkSqlOperator(
        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={}
    )

说明

所有DMS Airflow Operator均支持任务取消和自动重试等通用特性，详情请参见Airflow DMS Operator。