Migrating data from Kafka to MaxCompute

Step 1: Prepare Kafka data

You need to create test data in your Kafka cluster. First, configure the security group of the EMR cluster's header host to allow inbound traffic on TCP ports 22 and 9092. This ensures that you can log on to the header host and that MaxCompute and DataWorks can communicate with it.

1. Log on to the header host of the EMR cluster.

   1. Log on to the EMR console.

   2. In the top navigation bar, click Cluster Management.

   3. Find the target cluster and go to its details page.

   4. On the cluster details page, click the host list, find the IP address of the header host, and use an SSH client to connect to the host.

2. Create a test topic.

   Run the following command to create a test topic named testkafka:

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

3. Write test data.

   Run the following command to simulate a producer that writes data to the testkafka topic. Because Kafka is used for stream processing, you can continuously write data. To ensure a valid test result, write at least 10 messages.

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

   You can also open another SSH window and run the following command to simulate a consumer that verifies that the data was written to Kafka successfully. If successful, the data appears in the output.

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

Step 2: Create a target table in DataWorks

Create a target table in DataWorks to receive the data from Kafka.

1. Go to the Data Development page.

   1. Log on to the DataWorks console.

   2. In the left-side navigation pane, choose Data Development and O&M > Data Development.

   3. From the drop-down list, select the desired workspace and click Go to Data Studio.

2. Right-click a workflow and choose Create Table > MaxCompute > Table.

3. In the Create Table dialog box, enter a table name and click Create.

   Note

   • Table names must start with a letter and cannot contain Chinese characters or special characters.

   • If multiple MaxCompute data sources are bound in Data Studio, select the required MaxCompute Engine Instance.

4. On the table editing page, click DDL Statement.

5. In the DDL dialog box, enter the following statement and click Generate Table Schema.

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

   Each column corresponds to a built-in field from the DataWorks Data Integration Kafka reader:

   • __key__: the key of the message.

   • __value__: the full content of the message.

   • __partition__: the partition where the message resides.

   • __headers__: the header information of the message.

   • __offset__: the offset of the message.

   • __timestamp__: the timestamp of the message.

   You can also customize the column names. For more information, see Kafka Reader.

6. Click Deploy to Production Environment and OK.

Step 3: Synchronize data

1. Create an exclusive resource group for Data Integration.

   The public resource group in DataWorks does not fully support the Kafka plugin. You must use an exclusive resource group for Data Integration to synchronize data. For more information, see Use an exclusive resource group for Data Integration.

2. Create a Data Integration node.

   1. On the Data Studio page, right-click a workflow and choose Create Node > Data Integration > Batch Synchronization.

   2. In the Create Node dialog box, enter a Name for the node and click OK.

3. In the top navigation bar, choose icon.

4. In script mode, click icon.

5. Configure the script. The following code provides an example.

   {
       "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_source",// The name of the MaxCompute data source.
                   "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
           }
       }
   }

   You can run the kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --list command on the header host to find the consumer group name for the group.id parameter.

   • Command example

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

   • Sample output

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

   Using console-consumer-83505 as an example, you can run the kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --describe --group console-consumer-83505 command on the header host to determine the values of the beginOffset and endOffset parameters.

   • Command example:

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

   • Sample output:

     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. Configure the scheduling resource group.

   1. In the right-side navigation pane of the node configuration tab, click Properties.

   2. In the Resource Group section, set Resource Group for Scheduling to the exclusive resource group for Data Integration that you created.

      Note

      If you need to periodically write data from Kafka to MaxCompute, such as hourly, you can use the beginDateTime and endDateTime parameters. Configure the Data Integration task to run every hour with a one-hour data reading window. For more information, see Kafka Reader.

7. Click icon to run the code.

8. You can Runtime Log view the results.

Next steps

To verify that the data was synchronized, create a task in Data Studio to run an SQL statement against the target table. This topic uses select * from testkafka as an example. Perform the following steps: