最佳实践：创建离线同步任务OpenAPI基础实践 - 大数据开发治理平台 DataWorks

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

前提条件

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

使用限制

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

配置环境依赖及账号认证

配置Maven依赖。

打开Maven项目下的pom.xml文件，添加aliyun-java-sdk-core。

<dependency>
  <groupId>com.aliyun</groupId>
  <artifactId>aliyun-java-sdk-core</artifactId>
  <version>4.5.20</version>
</dependency>

打开Maven项目下的pom.xml文件，添加aliyun-java-sdk-dataworks-public。

<dependency>
  <groupId>com.aliyun</groupId>
  <artifactId>aliyun-java-sdk-dataworks-public</artifactId>
  <version>3.3.18</version>
</dependency>

客户端认证。
使用OpenAPI创建数据同步任务前，需要调用如下语句对登录阿里云的账号相关信息进行认证。如果阿里云的账号信息认证通过，则继续执行后续任务，如果认证不通过，则该调用会报错，您需要根据实际报错处理相关问题。
```
 DefaultProfile profile = DefaultProfile.getProfile(
                           "regionId",            //DataWorks工作空间所在的地域，例如cn-hangzhou。
                           "<yourAccessKeyId>",   //登录DataWorks工作空间的阿里云账号的AccessKey ID。
                           "<yourAccessSecret>"); //登录DataWorks工作空间的阿里云账号的AccessKey Secret。
 IAcsClient client = new DefaultAcsClient(profile);
```
您可以登录DataWorks控制台鼠标悬停至右上角的用户头像，单击AccessKey管理，进入AccessKey管理页面获取AccessKey ID和AccessKey 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 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);

        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来更新使用的独享资源组。更新后，您需要重新提交、发布同步任务，详情请参见配置流程。

Sample代码

POM 依赖

<dependency>
        <groupId>com.aliyun</groupId>
        <artifactId>aliyun-java-sdk-core</artifactId>
        <version>4.5.20</version>
</dependency>
<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>aliyun-java-sdk-dataworks-public</artifactId>
    <version>3.4.1</version>
 </dependency>

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 = "{\n" +
                "    \"type\": \"job\",\n" +
                "    \"version\": \"2.0\",\n" +
                "    \"steps\": [\n" +
                "        {\n" +
                "            \"stepType\": \"mysql\",\n" +
                "            \"parameter\": {\n" +
                "                \"envType\": 0,\n" +
                "                \"datasource\": \"mysql_autotest_dev\",\n" +
                "                \"column\": [\n" +
                "                    \"id\",\n" +
                "                    \"name\"\n" +
                "                ],\n" +
                "                \"connection\": [\n" +
                "                    {\n" +
                "                        \"datasource\": \"mysql_autotest_dev\",\n" +
                "                        \"table\": [\n" +
                "                            \"user\"\n" +
                "                        ]\n" +
                "                    }\n" +
                "                ],\n" +
                "                \"where\": \"\",\n" +
                "                \"splitPk\": \"\",\n" +
                "                \"encoding\": \"UTF-8\"\n" +
                "            },\n" +
                "            \"name\": \"Reader\",\n" +
                "            \"category\": \"reader\"\n" +
                "        },\n" +
                "        {\n" +
                "            \"stepType\": \"odps\",\n" +
                "            \"parameter\": {\n" +
                "                \"partition\": \"pt=${bizdate}\",\n" +
                "                \"truncate\": true,\n" +
                "                \"datasource\": \"odps_first\",\n" +
                "                \"envType\": 0,\n" +
                "                \"column\": [\n" +
                "                    \"id\",\n" +
                "                    \"name\"\n" +
                "                ],\n" +
                "                \"emptyAsNull\": false,\n" +
                "                \"tableComment\": \"null\",\n" +
                "                \"table\": \"user\"\n" +
                "            },\n" +
                "            \"name\": \"Writer\",\n" +
                "            \"category\": \"writer\"\n" +
                "        }\n" +
                "    ],\n" +
                "    \"setting\": {\n" +
                "        \"executeMode\": null,\n" +
                "        \"errorLimit\": {\n" +
                "            \"record\": \"\"\n" +
                "        },\n" +
                "        \"speed\": {\n" +
                "            \"concurrent\": 2,\n" +
                "            \"throttle\": false\n" +
                "        }\n" +
                "    },\n" +
                "    \"order\": {\n" +
                "        \"hops\": [\n" +
                "            {\n" +
                "                \"from\": \"Reader\",\n" +
                "                \"to\": \"Writer\"\n" +
                "            }\n" +
                "        ]\n" +
                "    }\n" +
                "}";
        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 IAcsClient client;

    public static void main(String[] args) throws Exception {
        String akId = "XX";
        String akSecret = "XX"; // 使用个人的ak
        String regionId = "cn-chengdu";
        IClientProfile profile = DefaultProfile.getProfile(regionId, akId, akSecret);
        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);   // 修改数据集成任务的调度属性
    }
}