Data Management:DMSSqlOperator

概要

DMSSqlOperator は、DMS で管理されているデータベースインスタンスに対して SQL ステートメントを実行し、その結果を取得します。

パラメーター

PollAsyncSQLExecuteResult

{
  "resultSetFileLink" : "https://xxxx", // SQLType が DQL の場合に返されます。
  "resultSetFileType" : "CSV", // DQL 操作の場合に CSV に設定されます。それ以外の場合は、このフィールドは空です。
  "resultSetOption" : {
    ""
  }, 
  "columnMetas": [
    {
      
    }
      
  ],
  "count" : 1,  // DQL 操作の場合に空ではありません。結果セット内の行数を示します。
  "affectRows" : 1, // DML および DQL 操作の場合、影響を受けた行数（例：更新された行数）を示します。
  "errorMessage" : "xxxx" // Status が FAILURE の場合に返されます。
}

例

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