Kafka数据迁移MaxCompute最佳实践

步骤一：准备Kafka数据

您需要在Kafka集群创建测试数据。为保证您可以顺利登录EMR集群Header主机，以及保证MaxCompute和DataWorks可以顺利和EMR集群Header主机通信，请您首先配置EMR集群Header主机安全组，放行TCP 22及TCP 9092端口。

1. 登录EMR集群Header主机地址。
   1. 进入EMR Hadoop控制台。
   2. 在顶部导航栏，单击集群管理。
   3. 在显示的页面，找到您需要创建测试数据的集群，进入集群详情页。
   4. 在集群详情页面，单击主机列表，确认EMR集群Header主机地址，并通过SSH连接远程登录。
2. 创建测试Topic。
   执行如下命令创建测试所使用的Topic testkafka。

   kafka-topics.sh --zookeeper emr-header-1:2181/kafka-1.0.1 --partitions 10 --replication-factor 3 --topic testkafka  --create

3. 写入测试数据。
   执行如下命令，可以模拟生产者向Topic testkafka中写入数据。由于Kafka用于处理流式数据，您可以持续不断的向其中写入数据。为保证测试结果，建议写入10条以上的数据。

   kafka-console-producer.sh --broker-list emr-header-1:9092 --topic testkafka

   您可以同时再打开一个SSH窗口，执行如下命令，模拟消费者验证数据是否已成功写入Kafka。当数据写入成功时，您可以看到已写入的数据。

   kafka-console-consumer.sh --bootstrap-server emr-header-1:9092 --topic testkafka --from-beginning

步骤二：在DataWorks上创建目标表

在DataWorks上创建目标表用以接收Kafka数据。

1. 进入数据开发页面。
   1. 登录DataWorks控制台。
   2. 在左侧导航栏，单击工作空间列表。
   3. 单击相应工作空间后的数据开发。
2. 右键单击业务流程，选择新建 > MaxCompute > 表。
3. 在弹出的新建表对话框中，填写表名，并单击提交。
   说明

   • 表名必须以字母开头，不能包含中文或特殊字符。
   • 如果绑定多个实例，则需要选择MaxCompute引擎实例。
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. 新建独享数据集成资源组。

   由于当前DataWorks的默认资源组无法完美支持Kafka插件，您需要使用独享数据集成资源组完成数据同步。详情请参见新增和使用独享数据集成资源组。

2. 新建数据集成节点。
   1. 进入数据开发页面，右键单击指定业务流程，选择新建 > 数据集成 > 离线同步。
   2. 在新建节点对话框中，输入节点名称，并单击提交。
3. 在顶部菜单栏上，单击图标。
4. 在脚本模式下，单击顶部菜单栏上的图标。
5. 配置脚本，示例代码如下。

   {
       "type": "job",
       "steps": [
           {
               "stepType": "kafka",
               "parameter": {
                   "server": "47.xxx.xxx.xxx:9092",
                   "kafkaConfig": {
                       "group.id": "console-consumer-83505"
                   },
                   "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
           }
       }
   }

   您可以通过在Header主机上使用kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --list命令查看group.id参数，及消费者的Group名称。

   • 命令示例

     kafka-consumer-groups.sh  --bootstrap-server emr-header-1:9092  --list

   • 返回结果

     _emr-client-metrics-handler-group
     console-consumer-69493
     console-consumer-83505
     console-consumer-21030
     console-consumer-45322
     console-consumer-14773

   以console-consumer-83505为例，您可以根据该参数在Header主机上使用kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --describe --group console-consumer-83505命令确认beginOffset及endOffset参数。

   • 命令示例。

     kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --describe --group console-consumer-83505

   • 返回结果

     TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG        CONSUMER-ID                                       HOST                           CLIENT-ID
     testkafka                      6          0               0               0          -                                                 -                              -
     test                           6          3               3               0          -                                                 -                              -
     testkafka                      0          0               0               0          -                                                 -                              -
     testkafka                      1          1               1               0          -                                                 -                              -
     testkafka                      5          0               0               0          -                                                 -                              -

6. 配置调度资源组。
   1. 在节点编辑页面的右侧导航栏，单击调度配置。
   2. 在资源属性区域，选择调度资源组为您创建的独享数据集成资源组。
      说明 如果您需要将Kafka的数据周期性（例如每小时）写入MaxCompute，您可以使用beginDateTime及endDateTime参数，设置数据读取的时间区间为1小时，然后每小时调度一次数据集成任务。详情请参见Kafka Reader。
7. 单击图标运行代码。
8. 您可以在运行日志查看运行结果。