DataHub通过数据集成批量导入数据 - 大数据开发治理平台 DataWorks

本文以Stream同步数据至DataHub的脚本模式为例，为您介绍如何通过数据集成导入离线DataHub数据。

前提条件

准备阿里云账号，并创建账号的访问密钥。详情请参见开通DataWorks服务。
开通MaxCompute，自动产生一个默认的MaxCompute数据源，并使用主账号登录DataWorks。
创建工作空间，您可以在工作空间中协作完成业务流程，共同维护数据和任务等。详情请参见创建工作空间。
说明
如果您需要通过子账号创建数据集成任务，请赋予其相应的权限。详情请参见创建RAM用户和空间级模块权限管控。

背景信息

数据集成是阿里云提供的数据同步平台。该平台具备可跨异构数据存储系统、可靠、安全、低成本、可弹性扩展等特点，可以为20多种数据源提供不同网络环境下的离线数据进出通道。本文以配置DataHub数据源为例，如果您需要使用其它类型的数据源配置同步任务，请参见支持的数据源及同步方案。

操作步骤

进入数据开发页面。
登录DataWorks控制台，切换至目标地域后，单击左侧导航栏的数据开发与治理 > 数据开发，在下拉框中选择对应工作空间后单击进入数据开发。
展开目标业务流程，右键单击数据集成，选择新建节点 > 离线同步。
在新建节点对话框中，输入名称并选择路径，单击确认。
说明
- 节点名称的长度不能超过128个字符。
- 此处的路径为创建的业务流程，具体操作请参见创建周期业务流程。
成功创建离线同步节点后，根据需要进行网络资源配置，单击下一步，选择工具栏中的图标。
单击提示对话框中的确认，即可进入脚本模式进行开发。
单击工具栏中的图标。
在导入模板对话框中，选择从来源端的Stream数据源同步至目标端的DataHub数据源的导入模板，单击确认。

导入模板后，根据自身需求编辑代码。

{
"type": "job",
"version": "1.0",
"configuration": {
 "setting": {
   "errorLimit": {
     "record": "0"
   },
   "speed": {
     "mbps": "1",//单位：MB/s。
     "concurrent": 1,//作业并发数。
     "throttle": false
   }
 },
 "reader": {
   "plugin": "stream",
   "parameter": {
     "column": [//源端列名。
       {
         "value": "field",//列属性。
         "type": "string"
       },
       {
         "value": true,
         "type": "bool"
       },
       {
         "value": "byte string",
         "type": "bytes"
       }
     ],
     "sliceRecordCount": "100000"
   }
 },
 "writer": {
   "plugin": "datahub",
   "parameter": {
     "datasource": "datahub",//数据源名。
     "topic": "xxxx",//Topic是DataHub订阅和发布的最小单位，您可以用Topic来表示一类或者一种流数据。
     "mode": "random",//随机写入。
     "shardId": "0",//Shard 表示对一个Topic进行数据传输的并发通道，每个Shard会有对应的ID。
     "maxCommitSize": 524288,//为了提高写出效率，待攒数据大小达到maxCommitSize大小（单位MB）时，批量提交到目的端。默认是1,048,576，即1MB数据。
     "maxRetryCount": 500
   }
 }
}
}

配置完成后，分别单击和图标。
说明
- DataHub仅支持以脚本模式导入数据。
- 如果需要选择新模板，请单击工具栏中的图标。一旦导入新模板，原有内容会被覆盖。
- 保存同步任务后，直接单击图标，任务会立刻运行。
  您也可以单击图标，提交同步任务至调度系统中。调度系统会按照配置属性，从第2天开始定时执行。