云原生大数据计算服务 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"}