阿里云大数据开发治理平台 DataWorks通过OpenAPI创建、修改、删除离线同步任务-云淘科技

本文为您介绍如何使用OpenAPI创建、修改、删除数据集成同步任务，同步来源端数据至去向端。

前提条件

已创建Maven项目，详情请参见创建Maven项目。
已创建业务流程，详情请参见创建周期业务流程。
已创建同步任务需要的数据源，详情请参见配置数据源。

使用限制

DataWorks当前仅支持使用OpenAPI创建数据集成离线同步任务。
调用CreateDISyncTask创建数据集成同步任务，仅支持使用脚本模式配置同步任务内容，详情请参见通过脚本模式配置离线同步任务。
DataWorks暂不支持使用OpenAPI创建业务流程，您需要使用现有的业务流程创建数据同步任务。

配置环境依赖及账号认证

配置Maven依赖。
1. 打开Maven项目下的pom.xml文件，添加aliyun-java-sdk-core。
```
  com.aliyun
  aliyun-java-sdk-core
  4.5.20
```
2. 打开Maven项目下的pom.xml文件，添加aliyun-java-sdk-dataworks-public。
```
  com.aliyun
  aliyun-java-sdk-dataworks-public
  3.3.18
```
3. 打开Maven项目，添加credentials-java（推荐使用Credentials最新版本）。
```
    com.aliyun
    credentials-java
    0.2.11
```
  说明通过阿里云Credentials工具托管AK，关于Credentials工具使用说明，详情请参见：身份验证配置。

客户端认证。使用OpenAPI创建数据同步任务前，需要调用如下语句对登录阿里云的账号相关信息进行认证。如果阿里云的账号信息认证通过，则继续执行后续任务，如果认证不通过，则该调用会报错，您需要根据实际报错处理相关问题。

// Please ensure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are set.https://help.aliyun.com/document_detail/378657.html        IClientProfile profile = DefaultProfile.getProfile("cn-shanghai", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));

配置流程

完成上述配置环境依赖及账号认证后，您可以通过OpenAPI调用相关接口，创建数据同步任务，同步来源端数据至去向端。配置流程如下：

创建数据集成任务。
配置任务的调度依赖。
提交数据集成任务。
发布同步任务至生产环境。

配置步骤

创建数据集成任务。调用CreateDISyncTask接口，创建数据集成任务。如下代码仅示例部分参数的配置，更多参数详情请参见CreateDISyncTask。

public void createFile() throws ClientException{
        CreateDISyncTaskRequest request = new CreateDISyncTaskRequest();
        request.setProjectId(181565L);
        request.setTaskType("DI_OFFLINE");
        request.setTaskContent("{\"type\":\"job\",\"version\":\"2.0\",\"steps\":[{\"stepType\":\"mysql\",\"parameter\":{\"envType\":1,\"datasource\":\"dh_mysql\",\"column\":[\"id\",\"name\"],\"tableComment\":\"same表comment\",\"connection\":[{\"datasource\":\"dh_mysql\",\"table\":[\"same\"]}],\"where\":\"\",\"splitPk\":\"id\",\"encoding\":\"UTF-8\"},\"name\":\"Reader\",\"category\":\"reader\"},{\"stepType\":\"odps\",\"parameter\":{\"partition\":\"pt=${bizdate}\",\"truncate\":true,\"datasource\":\"odps_first\",\"envType\":1,\"column\":[\"id\",\"name\"],\"emptyAsNull\":false,\"tableComment\":\"same表comment\",\"table\":\"same\"},\"name\":\"Writer\",\"category\":\"writer\"}],\"setting\":{\"errorLimit\":{\"record\":\"\"},\"speed\":{\"throttle\":false,\"concurrent\":2}},\"order\":{\"hops\":[{\"from\":\"Reader\",\"to\":\"Writer\"}]}}");
        request.setTaskParam("{\"FileFolderPath\":\"业务流程/new_biz/数据集成\",\"ResourceGroup\":\"S_res_group_280749521950784_1602767279794\"}");
        request.setTaskName("new_di_task_0607_1416");


        String regionId = "cn-hangzhou";
        // Please ensure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are set.https://help.aliyun.com/document_detail/378657.html
        IClientProfile profile = DefaultProfile.getProfile("cn-shanghai", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        DefaultProfile.addEndpoint("cn-hangzhou","dataworks-public","dataworks.cn-hangzhou.aliyuncs.com");
        IAcsClient client;

        client = new DefaultAcsClient(profile);

        CreateDISyncTaskResponse response1 = client.getAcsResponse(request);
        Gson gson1 = new Gson();

        System.out.println(gson1.toJson(response1));
}

调用UpdateFile接口，更新调度参数。请求参数配置详情如下表所示：


名称	类型	是否必选	示例值	描述
Action	String	是	UpdateFile	要执行的操作。
FileFolderPath	String	否	业务流程/第一个业务流程/数据集成/文件夹1/文件夹2	文件所在的路径。
ProjectId	Long	否	10000	DataWorks工作空间的ID。您可以登录DataWorks控制台，进入工作空间管理页面获取ID。
FileName	String	否	ods_user_info_d	文件的名称。您可以通过重新设置FileName的值来修改文件名称。例如，使用ListFiles接口查询目标目录下的文件ID，通过UpdateFile接口，输入查询的文件ID至FileId参数，并配置FileName的参数值，即可修改相应文件的名称。
FileDescription	String	否	这里是文件描述	文件的描述。
Content	String	否	SELECT “1”;	文件代码内容，不同代码类型（fileType）的文件，代码格式不同。您可以在运维中心，右键单击对应类型的任务，选择查看代码，查看具体的代码格式。
AutoRerunTimes	Integer	是	3	文件出错后，自动重跑的次数。
AutoRerunIntervalMillis	Integer	否	120000	出错自动重跑时间间隔，单位为毫秒。最大为1800000毫秒（30分钟）。该参数与DataWorks控制台中，数据开发任务的“调度配置>时间属性>出错自动重跑”的”重跑间隔“配置对应。控制台中“重跑间隔”的时间单位为分钟，请在调用时注意转换时间。
RerunMode	String	否	ALL_ALLOWED	重跑属性。取值如下： ALL_ALLOWED：运行成功或失败后皆可重跑。 FAILURE_ALLOWED：运行成功后不可重跑，运行失败后可以重跑。 ALL_DENIED：运行成功或失败皆不可重跑。该参数与DataWorks控制台中，数据开发任务的“调度配置>时间属性>重跑属性”配置内容对应。
Stop	Boolean	否	false	是否暂停调度，取值如下： true：暂停调度。 false：不暂停调度。该参数与DataWorks控制台中，数据开发任务的“调度配置>时间属性>调度类型”配置为”暂停调度“时对应。
ParaValue	String	否	x=a y=b z=c	调度参数。该参数与DataWorks控制台中，数据开发任务的“调度配置>参数”对应。您可以参考调度参数配置。
StartEffectDate	Long	否	936923400000	开始自动调度的毫秒时间戳。该参数与DataWorks控制台中，数据开发任务的“调度配置>时间属性>生效日期”配置的开始时间的毫秒时间戳对应。
EndEffectDate	Long	否	4155787800000	停止自动调度的时间戳，单位为毫秒。该参数与DataWorks控制台中，数据开发任务的“调度配置>时间属性>生效日期”配置的结束时间的毫秒时间戳对应。
CronExpress	String	否	00 00-59/5 1-23 * * ?	周期调度的cron表达式，该参数与DataWorks控制台中，数据开发任务的“调度配置>时间属性>cron表达式”对应。配置完“调度周期”及“定时调度时间”后，DataWorks会自动生成相应cron表达式。示例如下：每天凌晨5点30分定时调度：`00 30 05 * * ?`。每个小时的第15分钟定时调度：`00 15 * * * ?`。每隔十分钟调度一次：`00 00/10 * * * ?`。每天8点到17点，每隔十分钟调度一次：`00 00-59/10 8-23 * * * ?`。每月的1日0点20分自动调度：`00 20 00 1 * ?`。从1月1日0点10分开始，每过3个月调度一次：`00 10 00 1 1-12/3 ?`。每周二、周五的0点5分自动调度：`00 05 00 * * 2,5`。由于DataWorks调度系统的规则，cron表达式有以下限制：最短调度间隔时间为5分钟。每天最早调度时间为0点5分。
CycleType	String	否	NOT_DAY	调度周期的类型，包括NOT_DAY（分钟、小时）和DAY（日、周、月）。该参数与DataWorks控制台中，数据开发任务的“调度配置>时间属性>调度周期”对应。
DependentType	String	否	USER_DEFINE	依赖上一周期的方式。取值如下： SELF：依赖项选择本节点。 CHILD：依赖项选择一级子节点。 USER_DEFINE：依赖项选择其他节点。 NONE：未选择依赖项，即不会依赖上一周期。
DependentNodeIdList	String	否	5,10,15,20	当DependentType参数配置为USER_DEFINE时，用于设置当前文件具体依赖的节点ID。依赖多个节点时，使用英文逗号（,）分隔。该参数与DataWorks控制台中，数据开发任务的“调度配置>调度依赖”配置为“上一周期”后，依赖项选择”其他节点“时配置的内容对应。
InputList	String	否	project_root,project.file1,project.001_out	文件依赖的上游文件的输出名称。依赖多个输出时，使用英文逗号（,）分隔。该参数与DataWorks控制台中，数据开发任务的“调度配置>调度依赖“选择”同周期“时的”父节点输出名称”对应。
ProjectIdentifier	String	否	dw_project	DataWorks工作空间的名称。您可以登录DataWorks控制台，进入工作空间配置页面获取工作空间名称。该参数与ProjectId参数，二者必须设置其一，用来确定本次API调用操作的DataWorks工作空间。
FileId	Long	是	100000001	文件的ID。您可以调用ListFiles接口获取文件ID。
OutputList	String	否	dw_project.ods_user_info_d	文件的输出。该参数与DataWorks控制台中，数据开发任务的“调度配置>调度依赖“选择”同周期“时的”本节点的输出名称”对应。
ResourceGroupIdentifier	String	否	default_group	文件发布成任务后，任务执行时对应的资源组。您可以调用ListResourceGroups获取工作空间可用的资源组列表。
ConnectionName	String	否	odps_first	文件对应任务执行时，任务使用的数据源标识符。您可以调用ListDataSources获取可用的数据源列表。
Owner	String	否	18023848927592	文件所有者的用户ID。
AutoParsing	Boolean	否	true	文件是否开启自动解析功能。取值如下： true：文件会自动解析代码。 false：文件不会自动解析代码。该参数与DataWorks控制台中，数据开发任务的“调度配置>调度依赖”选择“同周期”时的“代码解析”对应。
SchedulerType	String	否	NORMAL	调度的类型，取值如下： NORMAL：正常调度任务。 MANUAL：手动任务，不会被日常调度，对应手动业务流程下的节点。 PAUSE：暂停任务。 SKIP：空跑任务，被日常调度，但启动调度时直接被置为成功。
AdvancedSettings	String	否	{“queue”:”default”,”SPARK_CONF”:”–conf spark.driver.memory=2g”}	任务的高级配置。该参数与DataWorks控制台中，EMR Spark Streaming和EMR Streaming SQL数据开发任务，编辑页面右侧导航栏的“高级设置“对应。当前仅EMR Spark Streaming和EMR Streaming SQL任务支持配置该参数，并且参数为JSON格式。
StartImmediately	Boolean	否	true	发布后是否立即启动。取值如下： true：发布后立即启动。 false：发布后暂不启动。该参数与DataWorks控制台中，EMR Spark Streaming和EMR Streaming SQL数据开发任务，编辑页面右侧导航栏的“配置>时间属性>启动方式“对应。
InputParameters	String	否	[{“ValueSource”: “project_001.first_node:bizdate_param”,”ParameterName”: “bizdate_input”}]	节点的上下文输入参数。参数为JSON格式，包含的字段可参考GetFile接口返回值中的InputContextParameterList参数结构。该参数与DataWorks控制台中，数据开发任务的“调度配置>节点上下文>本节点输入参数“对应。
OutputParameters	String	否	[{“Type”: 1,”Value”: “${bizdate}”,”ParameterName”: “bizdate_param”}]	节点的上下文输出参数。参数为JSON格式，包含的字段可参考GetFile接口返回值中的OutputContextParameterList参数结构。该参数与DataWorks控制台中，数据开发任务的“调度配置>节点上下文>本节点输出参数“对应。

提交数据集成任务。调用SubmitFile接口，提交数据集成任务至调度系统的开发环境。任务提交后，Response会返回deploymentId，您可以调用GetDeployment接口，通过deploymentId获取本次发布包的详细信息。

 public void submitFile() throws ClientException{
        SubmitFileRequest request = new SubmitFileRequest();
        request.setProjectId(78837L);
        request.setProjectIdentifier("zxy_8221431");
      // 此节点ID为创建节点时返回的ID，对应数据库File表的file_id。
        request.setFileId(501576542L);
        request.setComment("备注");
        SubmitFileResponse acsResponse = client.getAcsResponse(request);
      //DeploymentId为提交或发布的返回值。
      Long deploymentId = acsResponse.getData();
        log.info(acsResponse.toString());
}

上述代码仅示例部分参数的配置，更多参数详情请参见SubmitFile、获取发布包的详情。

发布同步任务到生产环境。调用DeployFile接口，发布数据集成同步任务至生产环境。说明仅标准模式的工作空间涉及执行该发布操作。

 public void deploy() throws ClientException{        DeployFileRequest request = new DeployFileRequest();        request.setProjectIdentifier("zxy_8221431");        request.setFileId(501576542L);        request.setComment("备注");        //NodeId和file_id二选一。NodeId的值为调度配置中基础属性的节点ID。        request.setNodeId(700004537241L);        DeployFileResponse acsResponse = client.getAcsResponse(request);         //DeploymentId为提交或发布的返回值。        Long deploymentId = acsResponse.getData();        log.info(acsResponse.getData().toString());    }

上述代码仅示例部分参数的配置，更多参数详情请参见DeployFile。

获取发布包详情。任务发布后，Response会返回deploymentId，您可以调用GetDeployment接口，通过deploymentId获取本次发布包的详细信息。当GetDeployment接口的返回参数Status取值为1时，则表示此次发布成功。

public void getDeployment() throws ClientException{        GetDeploymentRequest request = new GetDeploymentRequest();        request.setProjectId(78837L);        request.setProjectIdentifier("zxy_8221431");      //DeploymentId为提交或发布的返回值。调用GetDeployment接口，获取本次发布的具体情况。        request.setDeploymentId(2776067L);        GetDeploymentResponse acsResponse = client.getAcsResponse(request);        log.info(acsResponse.getData().toString());    }

上述代码仅示例部分参数的配置，更多参数详情请参见获取发布包的详情。

修改同步任务的相关配置

成功创建数据集成同步任务后，您可以调用UpdateDISyncTask接口更新任务的Content，或通过TaskParam来更新使用的独享资源组。更新后，您需要重新提交、发布同步任务，详情请参见配置流程。

删除同步任务

创建数据集成同步任务后，您可以调用DeleteFile接口删除同步任务。

Sample代码

POM 依赖


        com.aliyun
        aliyun-java-sdk-core
        4.5.20


    com.aliyun
    aliyun-java-sdk-dataworks-public
    3.4.1

Java Sdk调用代码

import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.IAcsClient;
import com.aliyuncs.dataworks_public.model.v20200518.*;
import com.aliyuncs.profile.DefaultProfile;
import com.aliyuncs.profile.IClientProfile;
import java.util.List;

public class createofflineTask {
    static Long createTask(String fileName) throws  Exception {
        Long projectId = 2043L;
        String taskType = "DI_OFFLINE";
        String taskContent = "{
" +
                "    \"type\": \"job\",
" +
                "    \"version\": \"2.0\",
" +
                "    \"steps\": [
" +
                "        {
" +
                "            \"stepType\": \"mysql\",
" +
                "            \"parameter\": {
" +
                "                \"envType\": 0,
" +
                "                \"datasource\": \"mysql_autotest_dev\",
" +
                "                \"column\": [
" +
                "                    \"id\",
" +
                "                    \"name\"
" +
                "                ],
" +
                "                \"connection\": [
" +
                "                    {
" +
                "                        \"datasource\": \"mysql_autotest_dev\",
" +
                "                        \"table\": [
" +
                "                            \"user\"
" +
                "                        ]
" +
                "                    }
" +
                "                ],
" +
                "                \"where\": \"\",
" +
                "                \"splitPk\": \"\",
" +
                "                \"encoding\": \"UTF-8\"
" +
                "            },
" +
                "            \"name\": \"Reader\",
" +
                "            \"category\": \"reader\"
" +
                "        },
" +
                "        {
" +
                "            \"stepType\": \"odps\",
" +
                "            \"parameter\": {
" +
                "                \"partition\": \"pt=${bizdate}\",
" +
                "                \"truncate\": true,
" +
                "                \"datasource\": \"odps_first\",
" +
                "                \"envType\": 0,
" +
                "                \"column\": [
" +
                "                    \"id\",
" +
                "                    \"name\"
" +
                "                ],
" +
                "                \"emptyAsNull\": false,
" +
                "                \"tableComment\": \"null\",
" +
                "                \"table\": \"user\"
" +
                "            },
" +
                "            \"name\": \"Writer\",
" +
                "            \"category\": \"writer\"
" +
                "        }
" +
                "    ],
" +
                "    \"setting\": {
" +
                "        \"executeMode\": null,
" +
                "        \"errorLimit\": {
" +
                "            \"record\": \"\"
" +
                "        },
" +
                "        \"speed\": {
" +
                "            \"concurrent\": 2,
" +
                "            \"throttle\": false
" +
                "        }
" +
                "    },
" +
                "    \"order\": {
" +
                "        \"hops\": [
" +
                "            {
" +
                "                \"from\": \"Reader\",
" +
                "                \"to\": \"Writer\"
" +
                "            }
" +
                "        ]
" +
                "    }
" +
                "}";
        CreateDISyncTaskRequest request = new CreateDISyncTaskRequest();
        request.setProjectId(projectId);
        request.setTaskType(taskType);
        request.setTaskContent(taskContent);
        request.setTaskName(fileName);
        request.setTaskParam("{\"FileFolderPath\":\"业务流程/自动化测试空间_勿动/数据集成\",\"ResourceGroup\":\"S_res_group_XXX\"}");
        // 使用数据集成独享资源组，
        CreateDISyncTaskResponse response1 = client.getAcsResponse(request);
        return response1.getData().getFileId();
    }

    public static void updateFile(Long fileId) throws Exception {
        UpdateFileRequest request = new UpdateFileRequest();
        request.setProjectId(2043L);
        request.setFileId(fileId);
        request.setAutoRerunTimes(3);
        request.setRerunMode("FAILURE_ALLOWED");
        request.setCronExpress("00 30 05 * * ?");
        request.setCycleType("DAY");
        request.setResourceGroupIdentifier("S_res_group_XXX");
        // 使用调度独享资源组
        request.setInputList("dataworks_di_autotest_root");

        request.setAutoParsing(true);
        request.setDependentNodeIdList("5,10,15,20");
        request.setDependentType("SELF");
        request.setStartEffectDate(0L);
        request.setEndEffectDate(4155787800000L);
        request.setFileDescription("description");
        request.setStop(false);
        request.setParaValue("x=a y=b z=c");
        request.setSchedulerType("NORMAL");
        request.setAutoRerunIntervalMillis(120000);
        UpdateFileResponse response1 = client.getAcsResponse(request);
    }
    static void deleteTask(Long fileId) throws Exception {
        DeleteFileRequest request = new DeleteFileRequest();
        Long projectId = 63845L;
        request.setProjectId(projectId);
        request.setFileId(fileId);


        String akId = "XXX";
        String akSecret = "XXXX";
        String regionId = "cn-hangzhou";
        IClientProfile profile = DefaultProfile.getProfile(regionId, akId, akSecret);
        DefaultProfile.addEndpoint("cn-hangzhou","dataworks-public","dataworks.cn-hangzhou.aliyuncs.com");
        IAcsClient client;

        client = new DefaultAcsClient(profile);

        DeleteFileResponse response1 = client.getAcsResponse(request);
        System.out.println(JSONObject.toJSONString(response1));
    }

    static IAcsClient client;

    public static void main(String[] args) throws Exception {
        String akId = "XX";
        // Please ensure that the environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET are set.https://help.aliyun.com/document_detail/378657.html
        IClientProfile profile = DefaultProfile.getProfile("cn-shanghai", System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        DefaultProfile.addEndpoint(regionId, "dataworks-public", "dataworks." + regionId + ".aliyuncs.com");

        client = new DefaultAcsClient(profile);
        String taskName = "offline_job_0930_1648";
        Long fileId = createTask(taskName); // 创建数据集成任务
        updateFile(fileId);   // 修改数据集成任务的调度属性
    }
}

内容没看懂？不太想学习？想快速解决？有偿解决：联系专家

阿里云企业补贴进行中：马上申请

腾讯云限时活动1折起，即将结束：马上收藏

 同尘科技为腾讯云授权服务中心。

购买腾讯云产品享受折上折，更有现金返利:同意关联，立享优惠

转转请注明出处：https://www.yunxiaoer.com/172139.html

阿里云大数据开发治理平台 DataWorks通过OpenAPI创建、修改、删除离线同步任务-云淘科技

前提条件

使用限制

配置环境依赖及账号认证

配置流程

配置步骤

修改同步任务的相关配置

删除同步任务

Sample代码

POM 依赖

Java Sdk调用代码

联系我们

400-800-8888

阿里云大数据开发治理平台 DataWorks通过OpenAPI创建、修改、删除离线同步任务-云淘科技

前提条件

使用限制

配置环境依赖及账号认证

配置流程

配置步骤

修改同步任务的相关配置

删除同步任务

Sample代码

POM 依赖

Java Sdk调用代码

相关推荐

阿里云RDS数据库DescribePGHbaConfig – 查询RDS PostgreSQL实例pg_hba.conf文件配置-云淘科技

阿里云RDS数据库DescribeDBProxyPerformance – 查询独享代理的性能数据-云淘科技

阿里云日志服务SLSOpenAPI集成概览-云淘科技

阿里云RDS数据库版本说明-云淘科技

阿里云RDS数据库GetDbProxyInstanceSsl – 查询独享代理连接地址SSL加密信息-云淘科技

联系我们

400-800-8888