DataHub通过数据集成批量导入数据

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

背景信息

数据集成是阿里云提供的数据同步平台。该平台具备可跨异构数据存储系统、可靠、安全、低成本、可弹性扩展等特点，可以为20多种数据源提供不同网络环境下的离线数据进出通道。数据源类型的详情请参见支持的数据源与读写插件。

本文以配置DataHub数据源为例，如果您需要使用其它类型的数据源配置同步任务，请参见配置Reader和配置Writer。

1. 进入数据集成页面。
   1. 登录DataWorks控制台。
   2. 在左侧导航栏，单击工作空间列表。
   3. 选择工作空间所在地域后，单击相应工作空间后的进入数据集成。
2. 在数据集成 > 首页页面，单击新建同步任务，进入数据开发页面。
3. 在新建节点对话框中，输入节点名称并选择目标文件夹，单击提交。
   说明

   • 节点名称的长度不能超过128个字符。
   • 此处的目标文件夹为创建的业务流程，具体操作请参见创建业务流程。
4. 成功创建离线同步节点后，单击工具栏中的图标。
5. 单击提示对话框中的确认，即可进入脚本模式进行开发。
6. 单击工具栏中的图标。
7. 在导入模板对话框中，选择从来源端的Stream数据源同步至目标端的DataHub数据源的导入模板，单击确认。
8. 导入模板后，根据自身需求编辑代码。

   {
   "type": "job",
   "version": "1.0",
   "configuration": {
    "setting": {
      "errorLimit": {
        "record": "0"
      },
      "speed": {
        "mbps": "1",
        "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
      }
    }
   }
   }

9. 配置完成后，分别单击和图标。
   说明

   • DataHub仅支持以脚本模式导入数据。
   • 如果需要选择新模板，请单击工具栏中的图标。一旦导入新模板，原有内容会被覆盖。
   • 保存同步任务后，直接单击图标，任务会立刻运行。

     您也可以单击图标，提交同步任务至调度系统中。调度系统会按照配置属性，从第2天开始定时执行。