DMSLindormSparkOperator - Data Management - Alibaba Cloud Documentation Center

Runs Spark SQL or JAR tasks on a Lindorm engine through DMS Airflow integration.

Overview

Runs tasks on the Lindorm Spark engine and supports two job types: SQL and JAR.

Parameters

Note

The instance, sql, region, and configs parameters support Jinja templates.

Parameter	Type	Required	Description
instance	string	Yes	The name of the connection (DBLink) to the Lindorm instance, as configured in DMS.
job_type	string	No	The job type. Valid values are: `sql` (default): Executes a Spark SQL statement. `jar`: Submits a JAR task.
sql	string	No	The Spark SQL statement to execute. This parameter is required if `job_type` is set to `sql`.
region	string	No	The Region ID of the Lindorm instance. Required for cross-region calls.
configs	dict	No	A configuration object. The following fields are required when `job_type` is set to `jar`: `mainClass`: The main class of the application. (Required) `mainResource`: The path to the JAR file in OSS, such as `oss://path/to/app.jar`. (Required) `args`: A list of arguments for the application. `appName`: The name of the application. `username`/`password`: The user credentials for the Lindorm instance. `configs`: A dictionary of Spark configuration properties, such as `{"spark.executor.memory": "4g"}`.
polling_interval	int	No	Interval in seconds for checking task status. Default: 10. Set to 0 or a negative number to submit the task without waiting for completion. Polling includes a built-in retry mechanism.

Examples

Note

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

SQL job

from airflow import DAG
from airflow.providers.alibabadms.cloud.operators.dms_lindorm_spark import DMSLindormSparkOperator

with DAG(
    "dms_lindorm_spark_sql",
) as dag:

    lindorm_sql = DMSLindormSparkOperator(
        task_id="lindorm_spark_sql",
        instance="my_lindorm_link",
        job_type="sql",
        sql="SELECT * FROM wide_table WHERE dt = '{{ ds }}' LIMIT 1000;",
        polling_interval=15,
        dag=dag,
    )

JAR job

from airflow import DAG
from airflow.providers.alibabadms.cloud.operators.dms_lindorm_spark import DMSLindormSparkOperator

with DAG(
    "dms_lindorm_spark_jar",
) as dag:

    lindorm_jar = DMSLindormSparkOperator(
        task_id="lindorm_spark_jar",
        instance="my_lindorm_link",
        job_type="jar",
        configs={
            "mainClass": "com.example.LindormETL",
            "mainResource": "oss://my-bucket/jars/lindorm-etl.jar",
            "args": ["--date", "{{ ds }}"],
            "configs": {
                "spark.executor.memory": "4g",
                "spark.executor.instances": "4",
            },
        },
        dag=dag,
    )

Note

All DMS Airflow operators support common features like task cancellation and automatic retry. For details, see the Airflow DMS Operator documentation.