MaxCompute：使用Kafka（離線與即時）

資料準備。

• 通過使用本地用戶端（odpscmd）串連或其他可以運行MaxCompute SQL的工具，建立目標MaxCompute表。

  CREATE TABLE IF NOT EXISTS table_text(
    topic STRING,
    `partition` BIGINT,
    `offset` BIGINT,
    key STRING,
    value STRING
  ) PARTITIONED BY (pt STRING);

• 建立Kafka資料。

  在$KAFKA_HOME/bin/目錄下，執行以下命令，建立Kafka Topic。以topic_text為例。

  sh kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_text

  執行以下命令，建立Kafka訊息。

  sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic topic_text --property parse.key=true
  >123    abc
  >456    edf

（可選）啟動Kafka-connector服務。具體操作，請參見配置並啟動Kafka-connector服務。

建立並配置odps-sink-connector.json檔案，並將odps-sink-connector.json檔案上傳至任意位置。本文以$KAFKA_HOME/config路徑為例。

odps-sink-connector.json檔案內容樣本如下，關於odps-sink-connector.json檔案詳情介紹，請參見配置並啟動Kafka-connector任務。

{
    "name": "odps-test-text",
    "config": {
      "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
      "tasks.max": "3",
      "topics": "topic_text",
      "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
      "project": "project_name",
      "schema":"default",
      "table": "table_text",
      "account_type": "ALIYUN",
      "access_id": "LTAI5tM2iHkTd4W69nof****",
      "access_key": "S0uZvwDYDa56WZ1tjVmA67z1YS****",
      "partition_window_type": "MINUTE",
      "mode":"VALUE",
      "format":"TEXT",
      "sink_pool_size":"150",
      "record_batch_size":"9000",
      "buffer_size_kb":"600000"
    }
  }

執行以下命令，啟動Kafka-connector資料轉送任務。

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors -d @$KAFKA_HOME/config/odps-sink-connector.json

結果驗證。

通過使用本地用戶端（odpscmd）串連或其他可以運行MaxCompute SQL的工具，執行如下命令，查詢資料寫入結果。

set odps.sql.allow.fullscan=true;
select * from table_text;

返回結果如下：

# 這裡由於我們odps-sink-connector.json設定檔中的mode值為VALUE，所以只保留value的內容，key欄位為NULL

+-------+------------+------------+-----+-------+----+
| topic | partition  | offset     | key | value | pt |
+-------+------------+------------+-----+-------+----+
| topic_text | 0      | 0          | NULL | abc   | 07-13-2023 21:13 |
| topic_text | 0      | 1          | NULL | edf   | 07-13-2023 21:13 |
+-------+------------+------------+-----+-------+----+

資料準備。

• 通過使用本地用戶端（odpscmd）串連或其他可以運行MaxCompute SQL的工具，建立目標MaxCompute表。

  CREATE TABLE IF NOT EXISTS table_csv(
    topic STRING,
    `partition` BIGINT,
    `offset` BIGINT,
    id BIGINT,
    name STRING,
    region STRING
  ) PARTITIONED BY (pt STRING);

• 建立Kafka資料。

  在$KAFKA_HOME/bin/目錄下，執行以下命令，建立Kafka Topic。以topic_csv為例。

  sh kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_csv

  執行以下命令，建立Kafka訊息。

  sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic topic_csv --property parse.key=true
  >123	1103,zhangsan,china
  >456	1104,lisi,usa

（可選）啟動Kafka-connector服務。具體操作，請參見配置並啟動Kafka-connector服務。

建立並配置odps-sink-connector.json檔案，並將odps-sink-connector.json檔案上傳至任意位置。本文以$KAFKA_HOME/config路徑為例。

odps-sink-connector.json檔案內容樣本如下，關於odps-sink-connector.json檔案詳情介紹，請參見配置並啟動Kafka-connector任務。

{
    "name": "odps-test-csv",
    "config": {
      "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
      "tasks.max": "3",
      "topics": "topic_csv",
      "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
      "project": "project_name",    
      "schema":"default",
      "table": "table_csv",
      "account_type": "ALIYUN",
      "access_id": "LTAI5tM2iHkTd4W69nof****",
      "access_key": "S0uZvwDYDa56WZ1tjVmA67z1YS****",
      "partition_window_type": "MINUTE",
      "format":"CSV",
      "mode":"VALUE",
      "sink_pool_size":"150",
      "record_batch_size":"9000",
      "buffer_size_kb":"600000"
    }
  }

執行以下命令，啟動Kafka-connector資料轉送任務。

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors -d @$KAFKA_HOME/config/odps-sink-connector.json

結果驗證。

通過使用本地用戶端（odpscmd）串連或其他可以運行MaxCompute SQL的工具，執行如下命令，查詢資料寫入結果。

set odps.sql.allow.fullscan=true;
select * from table_csv;

返回結果如下：

+-------+------------+------------+------------+------+--------+----+
| topic | partition  | offset     | id         | name | region | pt |
+-------+------------+------------+------------+------+--------+----+
| csv_test | 0       | 0          | 1103       | zhangsan | china  | 07-14-2023 00:10 |
| csv_test | 0       | 1          | 1104       | lisi | usa    | 07-14-2023 00:10 |
+-------+------------+------------+------------+------+--------+----+

資料準備。

• 通過使用本地用戶端（odpscmd）串連或其他可以運行MaxCompute SQL的工具，建立目標MaxCompute表。

  CREATE TABLE IF NOT EXISTS table_json(
    topic STRING,
    `partition` BIGINT,
    `offset` BIGINT,
    key STRING,
    value JSON
  ) PARTITIONED BY (pt STRING);

• 建立Kafka資料。

  在$KAFKA_HOME/bin/目錄下，執行以下命令，建立Kafka Topic。以topic_json為例。

  sh kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_json

  執行以下命令，建立Kafka訊息。

  sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic topic_json --property parse.key=true
  >123    {"id":123,"name":"json-1","region":"beijing"}                         
  >456    {"id":456,"name":"json-2","region":"hangzhou"}

（可選）啟動Kafka-connector服務。具體操作，請參見配置並啟動Kafka-connector服務。

建立並配置odps-sink-connector.json檔案，並將odps-sink-connector.json檔案上傳至任意位置。本文以$KAFKA_HOME/config路徑為例。

odps-sink-connector.json檔案內容樣本如下，關於odps-sink-connector.json檔案詳情介紹，請參見配置並啟動Kafka-connector任務。

{
    "name": "odps-test-json",
    "config": {
      "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
      "tasks.max": "3",
      "topics": "topic_json",
      "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
      "project": "project_name",    
      "schema":"default",
      "table": "table_json",
      "account_type": "ALIYUN",
      "access_id": "LTAI5tM2iHkTd4W69nof****",
      "access_key": "S0uZvwDYDa56WZ1tjVmA67z1YS****",
      "partition_window_type": "MINUTE",
      "mode":"VALUE",
      "format":"JSON",
      "sink_pool_size":"150",
      "record_batch_size":"9000",
      "buffer_size_kb":"600000"
    }
  }

執行以下命令，啟動Kafka-connector資料轉送任務。

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors -d @$KAFKA_HOME/config/odps-sink-connector.json

結果驗證。

通過使用本地用戶端（odpscmd）串連或其他可以運行MaxCompute SQL的工具，執行如下命令，查詢資料寫入結果。

set odps.sql.allow.fullscan=true;
select * from table_json;

返回結果如下：

# json 資料被成功寫入value欄位中
+-------+------------+------------+-----+-------+----+
| topic | partition  | offset     | key | value | pt |
+-------+------------+------------+-----+-------+----+
| Topic_json | 0      | 0          | NULL | {"id":123,"name":"json-1","region":"beijing"} | 07-14-2023 00:28 |
| Topic_json | 0      | 1          | NULL | {"id":456,"name":"json-2","region":"hangzhou"} | 07-14-2023 00:28 |
+-------+------------+------------+-----+-------+----+

資料準備。

• 通過使用本地用戶端（odpscmd）串連或其他可以運行MaxCompute SQL的工具，建立目標MaxCompute表。

  CREATE TABLE IF NOT EXISTS table_flatten(
    topic STRING,
    `partition` BIGINT,
    `offset` BIGINT,
    id BIGINT,
    name STRING,
    extendinfo JSON
  ) PARTITIONED BY (pt STRING);

• 建立Kafka資料。

  在$KAFKA_HOME/bin/目錄下，執行以下命令，建立Kafka Topic。以topic_flatten為例。

  ./kafka/bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_flatten

  執行以下命令，建立Kafka訊息。

  sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic topic_flatten --property parse.key=true
  >123  {"id":123,"name":"json-1","extendinfo":{"region":"beijing","sex":"M"}}                         
  >456  {"id":456,"name":"json-2","extendinfo":{"region":"hangzhou","sex":"W"}}

（可選）啟動Kafka-connector服務。具體操作，請參見配置並啟動Kafka-connector服務。

建立並配置odps-sink-connector.json檔案，並將odps-sink-connector.json檔案上傳至任意位置。本文以$KAFKA_HOME/config路徑為例。

odps-sink-connector.json檔案內容樣本如下，關於odps-sink-connector.json檔案詳情介紹，請參見配置並啟動Kafka-connector任務。

{
    "name": "odps-test-flatten",
    "config": {
      "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
      "tasks.max": "3",
      "topics": "topic_flatten",
      "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
      "project": "project_name",    
      "schema":"default",
      "table": "table_flatten",
      "account_type": "ALIYUN",
      "access_id": "LTAI5tM2iHkTd4W69nof****",
      "access_key": "S0uZvwDYDa56WZ1tjVmA67z1YS****",
      "partition_window_type": "MINUTE",
      "mode":"VALUE",
      "format":"FLATTEN",
      "sink_pool_size":"150",
      "record_batch_size":"9000",
      "buffer_size_kb":"600000"
    }
  }

執行以下命令，啟動Kafka-connector任務。

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors -d @$KAFKA_HOME/config/odps-sink-connector.json

結果驗證。

通過使用本地用戶端（odpscmd）串連或其他可以運行MaxCompute SQL的工具，執行如下命令，查詢資料寫入結果。

set odps.sql.allow.fullscan=true;
select * from table_flatten;

返回結果如下：

# json資料被解析寫入MaxCompute表中，且支援json巢狀型別exteninfo為JSON欄位
+-------+------------+--------+-----+------+------------+----+
| topic | partition  | offset | id  | name | extendinfo | pt |
+-------+------------+--------+-----+------+------------+----+
| topic_flatten | 0   | 0      | 123 | json-1 | {"sex":"M","region":"beijing"} | 07-14-2023 01:33 |
| topic_flatten | 0   | 1      | 456 | json-2 | {"sex":"W","region":"hangzhou"} | 07-14-2023 01:33 |
+-------+------------+--------+-----+------+------------+----+

資料準備。

• 通過使用本地用戶端（odpscmd）串連或其他可以運行MaxCompute SQL的工具，建立目標MaxCompute表。

  CREATE TABLE IF NOT EXISTS table_flatten(
    topic STRING,
    `partition` BIGINT,
    `offset` BIGINT,
    id BIGINT,
    name STRING,
    extendinfo JSON
  ) PARTITIONED BY (pt STRING);

• 建立Kafka資料。

  在$KAFKA_HOME/bin/目錄下，執行以下命令，建立Kafka Topic。

  • topic_abnormalTopic。

    sh kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic topic_abnormal

  • runtime_error異常訊息Topic。

    sh kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic runtime_error

    說明

    當資料寫入發生未知錯誤（通常是Kafka資料與MaxCompute表格式不匹配），異常資料會被寫入到runtime_errorTopic中。

  執行以下命令，建立Kafka訊息。

  以下訊息中，其中一條資料格式與目標MaxCompute表格式不匹配。

  sh kafka-console-producer.sh --bootstrap-server localhost:9092 --topic flatten_test --property parse.key=true
  
  >100  {"id":100,"name":"json-3","extendinfo":{"region":"beijing","gender":"M"}}                         
  >101  {"id":101,"name":"json-4","extendinfos":"null"}
  >102	{"id":102,"name":"json-5","extendinfo":{"region":"beijing","gender":"M"}} 

（可選）啟動Kafka-connector服務。具體操作，請參見配置並啟動Kafka-connector服務。

建立並配置odps-sink-connector.json檔案，並將odps-sink-connector.json檔案上傳至任意位置。本文以$KAFKA_HOME/config路徑為例。

odps-sink-connector.json檔案內容樣本如下，關於odps-sink-connector.json檔案詳情介紹，請參見配置並啟動Kafka-connector任務。

{
  "name": "odps-test-runtime-error",
  "config": {
    "connector.class": "com.aliyun.odps.kafka.connect.MaxComputeSinkConnector",
    "tasks.max": "3",
    "topics": "topic_abnormal",
    "endpoint": "http://service.cn-shanghai.maxcompute.aliyun.com/api",
    "project": "project_name",
    "schema":"default",
    "table": "test_flatten",
    "account_type": "ALIYUN",
    "access_id": "LTAI5tM2iHkTd4W69nof****",
    "access_key": "S0uZvwDYDa56WZ1tjVmA67z1YS****",
    "partition_window_type": "MINUTE",
    "mode":"VALUE",
    "format":"FLATTEN",
    "sink_pool_size":"150",
    "record_batch_size":"9000",
    "buffer_size_kb":"600000",
    "runtime.error.topic.name":"runtime_error",
    "runtime.error.topic.bootstrap.servers":"http://XXXX",
    "skip_error":"false"
  }
}

執行以下命令，啟動Kafka-connector任務。

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors -d @$KAFKA_HOME/config/odps-sink-connector.json

結果驗證。

• 查詢MaxCompute表資料

  通過使用本地用戶端（odpscmd）串連或其他可以運行MaxCompute SQL的工具，執行如下命令，查詢資料寫入結果。

  set odps.sql.allow.fullscan=true;
  select * from table_flatten;

  返回結果如下：

  # 我們看到最後兩條資料，因為設定了skip_error參數為true,所以id為101的資料沒有被寫入MaxCompute，且沒有block後面資料的寫入。
  +-------+------------+------------+------------+------+------------+----+
  | topic | partition  | offset     | id         | name | extendinfo | pt |
  +-------+------------+------------+------------+------+------------+----+
  | flatten_test | 0          | 0          | 123        | json-1 | {"gender":"M","region":"beijing"} | 07-14-2023 01:33 |
  | flatten_test | 0          | 1          | 456        | json-2 | {"gender":"W","region":"hangzhou"} | 07-14-2023 01:33 |
  | flatten_test | 0          | 0          | 123        | json-1 | {"gender":"M","region":"beijing"} | 07-14-2023 13:16 |
  | flatten_test | 0          | 1          | 456        | json-2 | {"gender":"W","region":"hangzhou"} | 07-14-2023 13:16 |
  | flatten_test | 0          | 2          | 100        | json-3 | {"gender":"M","region":"beijing"} | 07-14-2023 13:16 |
  | flatten_test | 0          | 4          | 102        | json-5 | {"gender":"M","region":"beijing"} | 07-14-2023 13:16 |
  +-------+------------+------------+------------+------+------------+----+

• 查詢runtime_errorTopic的訊息

  在$KAFKA_HOME/bin/目錄下，執行以下命令，查看訊息寫入結果。

  sh kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic runtime_error --from-beginning

  返回結果如下：

  # 異常資料被成功寫入runtime_error訊息佇列中
  {"id":101,"name":"json-4","extendinfos":"null"}