使用 Apache DolphinScheduler 调度 DLC 引擎提交任务

import json
import types
from tencentcloud.common import credential
from tencentcloud.common.profile.client_profile import ClientProfile
from tencentcloud.common.profile.http_profile import HttpProfile
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
from tencentcloud.dlc.v20210125 import dlc_client, models
import base64
import time
import sys


def create_task(sql):
    global err
    try:
        # 实例化一个认证对象，入参需要传入腾讯云账户 SecretId 和 SecretKey，此处还需注意密钥对的保密
        # 代码泄露可能会导致 SecretId 和 SecretKey 泄露，并威胁账号下所有资源的安全性。以下代码示例仅供参考，建议采用更安全的方式来使用密钥，请参见：https://cloud.tencent.com/document/product/1278/85305
        # 密钥可前往官网控制台 https://console.cloud.tencent.com/cam/capi 进行获取
        cred = credential.Credential("secretId", "secretKey")
        # 实例化一个http选项，可选的，没有特殊需求可以跳过
        httpProfile = HttpProfile()
        httpProfile.endpoint = "dlc.tencentcloudapi.com"

        # 实例化一个client选项，可选的，没有特殊需求可以跳过
        clientProfile = ClientProfile()
        clientProfile.httpProfile = httpProfile
        # 实例化要请求产品的client对象,clientProfile是可选的
        client = dlc_client.DlcClient(cred, "region", clientProfile)

        # 实例化一个请求对象,每个接口都会对应一个request对象
        req = models.CreateTasksRequest()
        base64sql = base64.b64encode(sql.encode('utf-8')).decode('utf-8')
        # print(base64sql)
        # print(base64sql)
        params = {
            "DatabaseName": "db",
            "Tasks": {
                "TaskType": "SQLTask",
                "FailureTolerance": "Terminate",
                "SQL": base64sql,
            },
            "DatasourceConnectionName": "DataLakeCatalog",
            "DataEngineName": "engineName"
        }
        req.from_json_string(json.dumps(params))

        # 返回的resp是一个CreateTasksResponse的实例，与请求对象对应
        resp = client.CreateTasks(req)
        # 输出json格式的字符串回包
        print(resp.to_json_string())

        return resp
    except TencentCloudSDKException as err:
        print(err)
      

if __name__ == "__main__":
    try:
        sql_file = "sql_file_path/dlc.sql"
        print(sql_file)
        with open(sql_file, 'r') as file:
            sqls = file.read()
            print(sqls)
        create_rsp = create_task(sqls)

    except Exception as main_err:
        print(main_err)



import jaydebeapi

jdbc_url = "jdbc:dlc:dlc.tencentcloudapi.com?task_type=SQLTask&datasource_connection_name=DataLakeCatalog®ion=ap-guangzhou&data_engine_name=public-engine"
user = "xx"
pwd = "xx"
driver = "com.tencent.cloud.dlc.jdbc.DlcDriver"
jar_file = '/opt/dolphinscheduler/libs/dlc-jdbc-2.2.3-jar-with-dependencies.jar'
sql = "select 1"
conn = jaydebeapi.connect(driver, jdbc_url, [user, pwd], jar_file)
curs = conn.cursor()
curs.execute(sql)
array_size = curs.arraysize.real
rowcount = curs.rowcount.real
print(array_size)
print(rowcount)
rows = curs.rowcount.real
if rows != 0:
    result = curs.fetchall()
    print(result)

curs.close()
conn.close()