通过数据开发实现数据增量迁移

DataArts Studio服务的DLF组件提供了一站式的大数据协同开发平台，借助DLF的在线脚本编辑、周期调度CDM的迁移作业，也可以实现增量数据迁移。

这里以DWS导入到OBS为例，介绍DLF配合CDM实现增量迁移的流程：

获取CDM作业的JSON
修改JSON
创建DLF作业

获取CDM作业的JSON

进入CDM主界面，创建一个DWS到OBS的表/文件迁移作业。
在CDM“作业管理”界面的“表/文件迁移”页签下，找到已创建的作业，单击作业操作列的“更多 > 查看作业JSON”，如图1所示。
您也可以使用其它已创建好的CDM作业JSON。

图1 查看作业JSON
作业JSON就是创建CDM作业的请求消息体模板，URL地址中[Endpoint]、{project_id}、{cluster_id}需要替换为您实际的信息：
- [Endpoint]：终端节点。
  终端节点（Endpoint）即调用API的请求地址，不同服务不同区域的终端节点不同。本服务的Endpoint可从终端节点Endpoint获取。
- {project_id}：项目ID。
- {cluster_id}：集群ID，可在CDM集管理界面，单击集群名称查看。

修改JSON

根据您的业务需要，可以修改JSON Body。这里以1天为周期，where子句作为抽取数据时的判断条件（一般使用时间字段来作为增量迁移时的判断条件），每天迁移昨天新增的数据。

修改where子句，增量某个时间段的数据：

     {
        "name": "fromJobConfig.whereClause",
        "value": "_timestamp >= '${startTime}' and _timestamp < '${currentTime}'"
     }

源端数据库是数据仓库服务DWS或者MySQL时，对于时间的判断可以写成以下两种：

_timestamp >= '2018-10-10 00:00:00' and _timestamp < '2018-10-11 00:00:00'
或者
_timestamp between '2018-10-10 00:00:00' and '2018-10-11 00:00:00'

如果源端数据库是Oracle，where子句应该写成：

_timestamp >= to_date (2018-10-10 00:00:00' , 'yyyy-mm-dd hh24:mi:ss' ) and _timestamp < to_date (2018-10-10 00:00:00' , 'yyyy-mm-dd hh24:mi:ss' )

每个周期的增量数据导入到不同的目录：

     {
        "name": "toJobConfig.outputDirectory",
        "value": "dws2obs/${currentTime}"
     }

作业名改成动态的，否则会因为作业重名而无法创建：

    "to-connector-name": "obs-connector",
    "from-link-name": "dws_link",
    "name": "dws2obs-${currentTime}"

如果需要修改更多参数，请参见《云数据迁移API参考》，这里修改后的JSON样例如下：

     
      
        
        
          {
  "jobs": [
    {
      "job_type": "NORMAL_JOB",
      "to-config-values": {
        "configs": [
          {
            "inputs": [
              {
                "name": "toJobConfig.bucketName",
                "value": "cdm-test"
              },
              {
                "name": "toJobConfig.outputDirectory",
                "value": "dws2obs/${currentTime}"
              },
              {
                "name": "toJobConfig.outputFormat",
                "value": "CSV_FILE"
              },
              {
                "name": "toJobConfig.fieldSeparator",
                "value": ","
              },
              {
                "name": "toJobConfig.writeToTempFile",
                "value": "false"
              },
              {
                "name": "toJobConfig.validateMD5",
                "value": "false"
              },
              {
                "name": "toJobConfig.encodeType",
                "value": "UTF-8"
              },
              {
                "name": "toJobConfig.duplicateFileOpType",
                "value": "REPLACE"
              },
              {
                "name": "toJobConfig.kmsEncryption",
                "value": "false"
              }
            ],
            "name": "toJobConfig"
          }
        ]
      },
      "from-config-values": {
        "configs": [
          {
            "inputs": [
              {
                "name": "fromJobConfig.schemaName",
                "value": "dws_database"
              },
              {
                "name": "fromJobConfig.tableName",
                "value": "dws_from"
              },
              {
                "name": "fromJobConfig.whereClause",
                "value": "_timestamp >= '${startTime}' and _timestamp < '${currentTime}'"
              },
              {
                "name": "fromJobConfig.columnList",
                "value": "_tiny&_small&_int&_integer&_bigint&_float&_double&_date&_timestamp&_char&_varchar&_text"
              }
            ],
            "name": "fromJobConfig"
          }
        ]
      },
      "from-connector-name": "generic-jdbc-connector",
      "to-link-name": "obs_link",
      "driver-config-values": {
        "configs": [
          {
            "inputs": [
              {
                "name": "throttlingConfig.numExtractors",
                "value": "1"
              },
              {
                "name": "throttlingConfig.submitToCluster",
                "value": "false"
              },
              {
                "name": "throttlingConfig.numLoaders",
                "value": "1"
              },
              {
                "name": "throttlingConfig.recordDirtyData",
                "value": "false"
              },
              {
                "name": "throttlingConfig.writeToLink",
                "value": "obs_link"
              }
            ],
            "name": "throttlingConfig"
          },
          {
            "inputs": [],
            "name": "jarConfig"
          },
          {
            "inputs": [],
            "name": "schedulerConfig"
          },
          {
            "inputs": [],
            "name": "transformConfig"
          },
          {
            "inputs": [],
            "name": "smnConfig"
          },
          {
            "inputs": [],
            "name": "retryJobConfig"
          }
        ]
      },
      "to-connector-name": "obs-connector",
      "from-link-name": "dws_link",
      "name": "dws2obs-${currentTime}"
    }
  ]
}

         

       

     
    

创建DLF作业

在DLF创建如图2所示的Rest Client节点数据开发作业，详细操作请参见《数据治理中心DataArts Studio 用户指南》的新建作业章节。
各节点与作业的配置详情请参见后续步骤。
图2 DLF作业
配置“创建作业”节点。
DLF通过Rest Client节点调用REST接口创建CDM迁移作业。配置Rest Client节点的属性如下：
1. 节点名称：您自定义名称，例如“创建CDM作业”。注意区分：在DLF作业中，CDM的迁移作业只是作为节点运行。
2. URL地址：配置为获取CDM作业的JSON中获取的URL，格式为https://{Endpoint}/cdm/v1.0/{project_id}/clusters/{cluster_id}/cdm/job。
3. HTTP方法：创建CDM作业的HTTP请求方法为“POST”。
4. 添加如下两个请求头：
  - Content-Type = application/json
  - X-Language = en-us
5. 请求消息体：输入修改JSON里面修改完成后的CDM作业JSON。
图3 创建CDM作业的节点属性
配置“运行作业”节点。
创建CDM作业的节点配置完后，还需要在后面添加运行CDM作业的REST节点，具体请参见《云数据迁移API参考》中的“启动作业”章节。配置RestAPI节点的属性如下：
1. 节点名称：运行作业。
2. URL地址：其中project_id、cluster_id和2. 配置“创建作业”节点中的保持一致，作业名需要配置为“dws2obs-${currentTime}”。格式为https://{Endpoint}/cdm/v1.0/{project_id}/clusters/{cluster_id}/cdm/job/{job_name}/start。
3. HTTP方法：运行CDM作业的HTTP请求方法为“PUT”。
4. 请求头：
  - Content-Type = application/json
  - X-Language = en-us
图4 运行CDM作业的节点属性
配置“等待作业运行完成”节点。
由于CDM作业是异步运行的，运行作业的REST请求返回200，不代表数据已经迁移成功。后续有计算作业依赖CDM的迁移作业时，需要一个RestAPI节点去周期判断迁移是否成功，如果CDM迁移成功，再去做计算操作。查询CDM迁移是否成功的API，具体请参见《云数据迁移API参考》中“查询作业状态”章节。
运行CDM作业的REST节点配置完成后，添加等待CDM作业完成节点，节点属性为：
1. 节点名称：等待作业运行完成。
2. URL地址：格式为https://{Endpoint}/cdm/v1.0/{project_id}/clusters/{cluster_id}/cdm/job/{job_name}/status。其中project_id、cluster_id和2. 配置“创建作业”节点中的保持一致，作业名需要配置为“dws2obs-${currentTime}”。
3. HTTP方法：查询CDM作业状态的HTTP请求方法为“GET”。
4. 请求头：
  - Content-Type = application/json
  - X-Language = en-us
5. 是否需要判断返回值：选择“YES”。
6. 返回值字段路径：配置为submissions[0].status。
7. 请求成功标志位：配置为SUCCEEDED。
8. 其他参数保持默认即可。
（可选）配置“删除作业运行完成”节点。
这里的删除作业可根据实际需要选择。由于DLF是通过周期创建CDM作业来实现增量迁移，因此会累积大量的作业在CDM集群上，所以可在迁移成功后，删除已经运行成功的作业。如果您需要删除，在查询CDM作业状态的节点后面，添加删除CDM作业的RestAPI节点即可，DLF会调用《云数据迁移API参考》中的“删除作业”接口。
删除CDM作业的节点属性为：
1. 节点名称：删除作业。
2. URL地址：格式为https://{Endpoint}/cdm/v1.0/{project_id}/clusters/{cluster_id}/cdm/job/{job_name}。其中project_id、cluster_id和2. 配置“创建作业”节点中的保持一致，作业名需要配置为“dws2obs-${currentTime}”。
3. HTTP方法：删除CDM作业的HTTP请求方法为“DELETE”。
4. 请求头：
  - Content-Type = application/json
  - X-Language = en-us
5. 其他参数保持默认即可。
图5 删除CDM作业节点配置
如果需要在迁移完后进行计算操作，可在后续添加各种计算节点，完成数据计算。
配置DLF作业参数。
1. 配置DLF作业参数，如图6所示。
  - startTime = $getTaskPlanTime(plantime,@@yyyyMMddHHmmss@@,-24*60*60)
  - currentTime = $getTaskPlanTime(plantime,@@yyyyMMdd-HHmm@@,0)
  图6 DLF作业参数配置
2. 保存DLF作业后，选择“调度配置 > 周期调度”，调度周期配置为1天。
  这样，DLF配合CDM就实现了每天迁移昨天新增的数据。