将消息队列for Apache Kafka的数据迁移至MaxCompute

背景信息

MaxCompute是一种大数据计算服务，能提供快速、完全托管免运维的 EB 级云数据仓库解决方案。

DataWorks是基于MaxCompute计算和存储，提供工作流可视化开发、调度运维托管的一站式海量数据离线加工分析平台。在数加（一站式大数据平台）中，DataWorks 控制台即为 MaxCompute 控制台。MaxCompute 和 DataWorks 一起向用户提供完善的 ETL 和数仓管理能力，以及 SQL、MR、Graph 等多种经典的分布式计算模型，能够更快速地解决用户海量数据计算问题，有效降低企业成本，保障数据安全。

本教程旨在帮助您使用 DataWorks，将消息队列for Apache Kafka中的数据导入至 MaxCompute，来进一步探索大数据的价值。

步骤一：准备 Kafka 数据

1. 登录消息队列 for Apache Kafka 控制台创建 Topic 和 Consumer Group，分别命名为 testkafka 和 console-consumer。具体步骤参见步骤三：创建资源。本示例中，Consumer Group console-consumer 将用于消费 Topic testkafka 中的数据。
2. 向 Topic testkafka 中写入数据。由于消息队列 for Apache Kafka 用于处理流式数据，您可以持续不断地向其中写入数据。为保证测试结果，建议您写入 10 条以上的数据。您可以直接在控制台使用发送消息功能来写入数据，也可以使用消息队列 for Apache Kafka 的 SDK 收发消息。详情参见使用 SDK 收发消息。
3. 为验证写入数据生效，您可以在控制台查询消息，看到之前写入 Topic 中的数据。

步骤二：创建 DataWorks 表

您需创建 DataWorks 表，以保证大数据计算服务 MaxCompute 可以顺利接收消息队列 for Apache Kafka 数据。本例中为测试便利，使用非分区表。

1. 登录 DataWorks 控制台，在工作空间区域，单击目标工作空间的进入数据开发。
2. 在左侧导航栏单击表管理，然后单击新建图标。
   说明 DataWorks中创建表的更多详情，请参见表管理。
3. 在新建表对话框，输入表名 testkafka，然后单击提交。
4. 在创建的表页面，单击 DDL模式。
5. 在 DDL模式对话框，输入以下建表语句，单击生成表结构。

   CREATE TABLE `testkafka` (
       `key` string,
       `value` string,
       `partition1` string,
       `timestamp1` string,
       `offset` string,
       `t123` string,
       `event_id` string,
       `tag` string
   ) ;

   建表语句中的每一列对应 DataWorks 数据集成 Kafka Reader 的默认列。

   • key：表示消息的 Key。
   • value：表示消息的完整内容 。
   • partition：表示当前消息所在分区。
   • headers：表示当前消息 headers 信息。
   • offset：表示当前消息的偏移量。
   • timestamp：表示当前消息的时间戳。

   您还可以自主命名，详情参见配置 Kafka Reader。

6. 单击提交到生产环境。

步骤三：同步数据

1. 新增任务资源。此处创建的 ECS 实例将用以完成数据同步任务。
2. 登录 DataWorks 控制台，在工作空间区域，单击目标工作空间的进入数据开发。
3. 在左侧导航栏，选择数据开发 > 业务流程 > 数据迁移。
4. 右键选择数据集成 > 新建数据集成节点 > 数据同步。
5. 在新建节点对话框，输入节点名称（即数据同步任务名称），然后单击提交。
6. 在创建的节点页面，选择数据来源的数据源为 Kafka ，选择数据去向的数据源为ODPS，选择您在步骤二：创建 DataWorks 表中创建的表。完成上述配置后，请单击框中的按钮，转换为脚本模式，如下图所示。
7. 配置脚本，示例如下。

   {
       "type": "job",
       "steps": [
           {
               "stepType": "kafka",
               "parameter": {
                   "server": "47.xxx.xxx.xxx:9092",
                   "kafkaConfig": {
                       "group.id": "console-consumer"
                   },
                   "valueType": "ByteArray",
                   "column": [
                       "__key__",
                       "__value__",
                       "__partition__",
                       "__timestamp__",
                       "__offset__",
                       "'123'",
                       "event_id",
                       "tag.desc"
                   ],
                   "topic": "testkafka",
                   "keyType": "ByteArray",
                   "waitTime": "10",
                   "beginOffset": "0",
                   "endOffset": "3"
               },
               "name": "Reader",
               "category": "reader"
           },
           {
               "stepType": "odps",
               "parameter": {
                   "partition": "",
                   "truncate": true,
                   "compress": false,
                   "datasource": "odps_first",
                   "column": [
                       "key",
                       "value",
                       "partition1",
                       "timestamp1",
                       "offset",
                       "t123",
                       "event_id",
                       "tag"
                   ],
                   "emptyAsNull": false,
                   "table": "testkafka"
               },
               "name": "Writer",
               "category": "writer"
           }
       ],
       "version": "2.0",
       "order": {
           "hops": [
               {
                   "from": "Reader",
                   "to": "Writer"
               }
           ]
       },
       "setting": {
           "errorLimit": {
               "record": ""
           },
           "speed": {
               "throttle": false,
               "concurrent": 1
           }
       }
   }

8. 在脚本页面，单击配置任务资源组，选择步骤 1 中创建的自定义资源组，然后单击运行图标。