DMSSqlOperator - Data Management - Alibaba Cloud Documentation Center

This document describes how to configure the DMSSqlOperator.

Overview

The DMSSqlOperator executes an SQL statement on a DMS-managed database instance and retrieves the results.

Parameters

Note

The parameters instance, database, and sql support Jinja templates.

Parameter	Type	Required	Description
instance	string	Yes	The DBLink connection name for the DMS-managed database instance.
database	string	Yes	The database name.
sql	string	Yes	The SQL statement to execute. Note To run multiple SQL statements, separate them with a semicolon (;).
csv_null_replace_str	string	No	Used to replace `null` values in the `resultset`. The default value is the string "null".
callback	function	No	A callback function to process the SQL operation's result. The input parameter is PollAsyncSQLExecuteResult. Note This parameter takes effect only when `polling_interval` is greater than 0.
polling_interval	int	No	The polling interval, in seconds, for the execution result. The default is 10. If you set this value to 0 or less, the task is submitted without waiting for the result. The status check includes a built-in retry mechanism.
show_return_value_in_logs	bool	No	Specifies whether to write the return value of the callback function to the log. Defaults to `True`.

PollAsyncSQLExecuteResult

Parameter	Type	Description
Status	string	The execution status of the SQL operation. Valid values are: WAITING RUNNING SUCCESS FAILURE
SQLType	string	The type of SQL operation. Valid values are: DDL DML DQL UNKNOWN Note Indicates that the SQL type could not be parsed.
ResultType	string	The result type of the SQL operation. Note An empty value indicates the SQL operation has not completed. PLAINTEXT: The content of `ResultContent`, which is typically DDL and DML information. FILE: The result is a file that must be downloaded from OSS. The default format is CSV, which is typical for DQL operations.
ResultContent	JSON	The result of the SQL operation. Note When you download a file using the `resultSetFileLink` link, you must specify the request header "x-oss-range-behavior:standard". Otherwise, signature verification will fail. { "resultSetFileLink" : "https://xxxx", // Returned when SQLType is DQL. "resultSetFileType" : "CSV", // Set to CSV for DQL operations. Otherwise, this field is empty. "resultSetOption" : { "" }, "columnMetas": [ { } ], "count" : 1, // Not empty for DQL operations. Indicates the number of rows in the result set. "affectRows" : 1, // For DML and DQL operations, indicates the number of affected rows, for example, the number of updated rows. "errorMessage" : "xxxx" // Returned when Status is FAILURE. }

Example

Note

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

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

from airflow.providers.alibabadms.cloud.operators.dms_sql import DMSSqlOperator

import json
import requests

def callback(result):
    print(f"result: {result}")
    if 'Data' in result and 'ResultContent' in result['Data']:
        link = result['Data']['ResultContent']['ResultSetFileLink']
        print(f"get link: {link}")
        http_res = requests.get(link, headers={
            "x-oss-range-behavior": "standard"
        })
        print(f"link res: {http_res.text}")


with DAG(
    "dms_sql_dblink",
    params={
    },
) as dag:

    sql_operator = DMSSqlOperator(
        task_id="sql_test_dblink",
        instance="dblink_90",
        database="student_db",
        sql="show databases;show tables;",
        csv_null_replace_str="null",
        callback=callback,
        polling_interval=5,
        dag=dag
    )

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

    sql_operator >> run_this_last

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

Note

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