步驟一：準(zhǔn)備Kafka數(shù)據(jù)

您需要在Kafka集群創(chuàng)建測(cè)試數(shù)據(jù)。為保證您可以順利登錄EMR集群Header主機(jī)，以及保證MaxCompute和DataWorks可以順利和EMR集群Header主機(jī)通信，請(qǐng)您首先配置EMR集群Header主機(jī)安全組，放行TCP 22及TCP 9092端口。

1. 登錄EMR集群Header主機(jī)地址。

   1. 進(jìn)入EMR Hadoop控制臺(tái)。

   2. 在頂部導(dǎo)航欄，單擊集群管理。

   3. 在顯示的頁(yè)面，找到您需要?jiǎng)?chuàng)建測(cè)試數(shù)據(jù)的集群，進(jìn)入集群詳情頁(yè)。

   4. 在集群詳情頁(yè)面，單擊主機(jī)列表，確認(rèn)EMR集群Header主機(jī)地址，并通過(guò)SSH鏈接遠(yuǎn)程登錄。

2. 創(chuàng)建測(cè)試Topic。

   執(zhí)行如下命令創(chuàng)建測(cè)試所使用的Topic testkafka。

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

3. 寫(xiě)入測(cè)試數(shù)據(jù)。

   執(zhí)行如下命令，可以模擬生產(chǎn)者向Topic testkafka中寫(xiě)入數(shù)據(jù)。由于Kafka用于處理流式數(shù)據(jù)，您可以持續(xù)不斷地向其中寫(xiě)入數(shù)據(jù)。為保證測(cè)試結(jié)果，建議寫(xiě)入10條以上的數(shù)據(jù)。

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

   您可以同時(shí)再打開(kāi)一個(gè)SSH窗口，執(zhí)行如下命令，模擬消費(fèi)者驗(yàn)證數(shù)據(jù)是否已成功寫(xiě)入Kafka。當(dāng)數(shù)據(jù)寫(xiě)入成功時(shí)，您可以看到已寫(xiě)入的數(shù)據(jù)。

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

步驟二：在DataWorks上創(chuàng)建目標(biāo)表

在DataWorks上創(chuàng)建目標(biāo)表用以接收Kafka數(shù)據(jù)。

1. 進(jìn)入數(shù)據(jù)開(kāi)發(fā)頁(yè)面。

   1. 登錄DataWorks控制臺(tái)。

   2. 單擊左側(cè)導(dǎo)航欄數(shù)據(jù)開(kāi)發(fā)與治理 > 數(shù)據(jù)開(kāi)發(fā)。

   3. 在下拉框中選擇對(duì)應(yīng)工作空間后單擊進(jìn)入數(shù)據(jù)開(kāi)發(fā)。

2. 右鍵單擊業(yè)務(wù)流程，選擇新建 > MaxCompute > 表。

3. 在彈出的新建表對(duì)話框中，填寫(xiě)表名稱，并單擊新建。

   說(shuō)明

   • 表名必須以字母開(kāi)頭，不能包含中文或特殊字符。

   • 如果在數(shù)據(jù)開(kāi)發(fā)中綁定多個(gè)MaxCompute數(shù)據(jù)源，則按需選擇MaxCompute引擎實(shí)例。

4. 在表的編輯頁(yè)面，單擊DDL模式。

5. 在DDL對(duì)話框中，輸入如下建表語(yǔ)句，單擊生成表結(jié)構(gòu)。

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

   其中的每一列，對(duì)應(yīng)于DataWorks數(shù)據(jù)集成Kafka Reader的默認(rèn)列：

   • __key__表示消息的key。

   • __value__表示消息的完整內(nèi)容 。

   • __partition__表示當(dāng)前消息所在分區(qū)。

   • __headers__表示當(dāng)前消息headers信息。

   • __offset__表示當(dāng)前消息的偏移量。

   • __timestamp__表示當(dāng)前消息的時(shí)間戳。

   您還可以自主命名，詳情參見(jiàn)Kafka Reader。

6. 單擊提交到生產(chǎn)環(huán)境并確認(rèn)。

步驟三：同步數(shù)據(jù)

1. 新建獨(dú)享數(shù)據(jù)集成資源組。

   由于當(dāng)前DataWorks的公共資源組無(wú)法完美支持Kafka插件，您需要使用獨(dú)享數(shù)據(jù)集成資源組完成數(shù)據(jù)同步。詳情請(qǐng)參見(jiàn)新增和使用獨(dú)享數(shù)據(jù)集成資源組。

2. 新建數(shù)據(jù)集成節(jié)點(diǎn)。

   1. 進(jìn)入數(shù)據(jù)開(kāi)發(fā)頁(yè)面，右鍵單擊指定業(yè)務(wù)流程，選擇新建節(jié)點(diǎn) > 數(shù)據(jù)集成 > 離線同步。

   2. 在新建節(jié)點(diǎn)對(duì)話框中，輸入節(jié)點(diǎn)名稱，并單擊確認(rèn)。

3. 在頂部菜單欄上，單擊圖標(biāo)。

4. 在腳本模式下，單擊頂部菜單欄上的圖標(biāo)。

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_source",// MaxCompute數(shù)據(jù)源名稱
                   "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
           }
       }
   }

   您可以通過(guò)在Header主機(jī)上使用kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --list命令查看group.id參數(shù)，及消費(fèi)者的Group名稱。

   • 命令示例

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

   • 返回結(jié)果

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

   以console-consumer-83505為例，您可以根據(jù)該參數(shù)在Header主機(jī)上使用kafka-consumer-groups.sh --bootstrap-server emr-header-1:9092 --describe --group console-consumer-83505命令確認(rèn)beginOffset及endOffset參數(shù)。

   • 命令示例。

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

   • 返回結(jié)果

     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. 配置調(diào)度資源組。

   1. 在節(jié)點(diǎn)編輯頁(yè)面的右側(cè)導(dǎo)航欄，單擊調(diào)度配置。

   2. 在資源屬性區(qū)域，選擇調(diào)度資源組為您創(chuàng)建的獨(dú)享數(shù)據(jù)集成資源組。

      說(shuō)明

      如果您需要將Kafka的數(shù)據(jù)周期性（例如每小時(shí)）寫(xiě)入MaxCompute，您可以使用beginDateTime及endDateTime參數(shù)，設(shè)置數(shù)據(jù)讀取的時(shí)間區(qū)間為1小時(shí)，然后每小時(shí)調(diào)度一次數(shù)據(jù)集成任務(wù)。詳情請(qǐng)參見(jiàn)Kafka Reader。

7. 單擊圖標(biāo)運(yùn)行代碼。

8. 您可以在運(yùn)行日志查看運(yùn)行結(jié)果。

后續(xù)步驟

您可以新建一個(gè)數(shù)據(jù)開(kāi)發(fā)任務(wù)運(yùn)行SQL語(yǔ)句，查看當(dāng)前表中是否已存在從云消息隊(duì)列 Kafka 版同步過(guò)來(lái)的數(shù)據(jù)。本文以select * from testkafka為例，具體步驟如下：