MaxCompute:Import Kafka data to MaxCompute in offline or real-time mode

Prepare data.

• Create a MaxCompute table by using the MaxCompute client (odpscmd) or another tool that can run MaxCompute SQL.

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

• Create Kafka data.

  In the $KAFKA_HOME/bin/ directory, run the following command to create a Kafka topic. In this example, a Kafka topic named topic_text is created.

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

  Run the following command to create Kafka messages:

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

(Optional) Start the Kafka-connector service. For more information, see Configure and start the Kafka-connector service.

Create and configure the odps-sink-connector.json file, and upload the odps-sink-connector.json file to any location. In this example, the odps-sink-connector.json file is uploaded to the $KAFKA_HOME/config directory.

The following code shows the content of the odps-sink-connector.json file. For more information about the odps-sink-connector.json file, see Configure and start a Kafka-connector task.

{
    "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"
    }
  }

Run the following command to start the Kafka-connector task:

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

Verify the result.

Run the following commands on the MaxCompute client (odpscmd) or another tool that can run MaxCompute SQL to query the data write result:

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

The following result is returned:

# The mode value in the odps-sink-connector.json configuration file is VALUE. Therefore, only the value is retained and the key field is 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 |
+-------+------------+------------+-----+-------+----+

Prepare data.

• Create a destination MaxCompute table by using the MaxCompute client (odpscmd) or another tool that can run MaxCompute SQL.

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

• Create Kafka data.

  In the $KAFKA_HOME/bin/ directory, run the following command to create a Kafka topic. In this example, a Kafka topic named topic_csv is created.

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

  Run the following command to create Kafka messages:

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

(Optional) Start the Kafka-connector service. For more information, see Configure and start the Kafka-connector service.

Create and configure the odps-sink-connector.json file, and upload the odps-sink-connector.json file to any location. In this example, the odps-sink-connector.json file is uploaded to the $KAFKA_HOME/config directory.

The following code shows the content of the odps-sink-connector.json file. For more information about the odps-sink-connector.json file, see Configure and start a Kafka-connector task.

{
    "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"
    }
  }

Run the following command to start the Kafka-connector task:

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

Verify the result.

Run the following commands on the MaxCompute client (odpscmd) or another tool that can run MaxCompute SQL to query the data write result:

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

The following result is returned:

+-------+------------+------------+------------+------+--------+----+
| 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 |
+-------+------------+------------+------------+------+--------+----+

Prepare data.

• Create a destination MaxCompute table by using the MaxCompute client (odpscmd) or another tool that can run MaxCompute SQL.

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

• Create Kafka data.

  In the $KAFKA_HOME/bin/ directory, run the following command to create a Kafka topic. In this example, a topic named topic_json is created.

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

  Run the following command to create Kafka messages:

  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"}

(Optional) Start the Kafka-connector service. For more information, see Configure and start the Kafka-connector service.

Create and configure the odps-sink-connector.json file, and upload the odps-sink-connector.json file to any location. In this example, the odps-sink-connector.json file is uploaded to the $KAFKA_HOME/config directory.

The following code shows the content of the odps-sink-connector.json file. For more information about the odps-sink-connector.json file, see Configure and start a Kafka-connector task.

{
    "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"
    }
  }

Run the following command to start the Kafka-connector task:

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

Verify the result.

Run the following commands on the MaxCompute client (odpscmd) or another tool that can run MaxCompute SQL to query the data write result:

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

The following result is returned:

# Write JSON data to the value field.
+-------+------------+------------+-----+-------+----+
| 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 |
+-------+------------+------------+-----+-------+----+

Prepare data.

• Create a destination MaxCompute table by using the MaxCompute client (odpscmd) or another tool that can run MaxCompute SQL.

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

• Create Kafka data.

  In the $KAFKA_HOME/bin/ directory, run the following command to create a Kafka topic. In this example, a topic named topic_flatten is created.

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

  Run the following command to create Kafka messages:

  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"}}

(Optional) Start the Kafka-connector service. For more information, see Configure and start the Kafka-connector service.

Create and configure the odps-sink-connector.json file, and upload the odps-sink-connector.json file to any location. In this example, the odps-sink-connector.json file is uploaded to the $KAFKA_HOME/config directory.

The following code shows the content of the odps-sink-connector.json file. For more information about the odps-sink-connector.json file, see Configure and start a Kafka-connector task.

{
    "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"
    }
  }

Run the following command to start the Kafka-connector task:

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

Verify the result.

Run the following commands on the MaxCompute client (odpscmd) or another tool that can run MaxCompute SQL to query the data write result:

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

The following result is returned:

# JSON data is parsed and written to the MaxCompute table. The exteninfo field in the nested JSON format can be a JSON field.
+-------+------------+--------+-----+------+------------+----+
| 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 |
+-------+------------+--------+-----+------+------------+----+

Prepare data.

• Create a destination MaxCompute table by using the MaxCompute client (odpscmd) or another tool that can run MaxCompute SQL.

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

• Create Kafka data.

  In the $KAFKA_HOME/bin/ directory, run commands to create the following Kafka topics:

  • topic_abnormal

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

  • runtime_error

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

    Note

    If an unknown error occurs when data is written, the abnormal data is written to the runtime_error topic. In most cases, an unknown error occurs because the format of Kafka data is not consistent with the format of the MaxCompute table.

  Run the following command to create Kafka messages:

  In the following messages, the data format of one message is not the same as the format of the MaxCompute table.

  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"}} 

(Optional) Start the Kafka-connector service. For more information, see Configure and start the Kafka-connector service.

Create and configure the odps-sink-connector.json file, and upload the odps-sink-connector.json file to any location. In this example, the odps-sink-connector.json file is uploaded to the $KAFKA_HOME/config directory.

The following code shows the content of the odps-sink-connector.json file. For more information about the odps-sink-connector.json file, see Configure and start a Kafka-connector task.

{
  "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"
  }
}

Run the following command to start the Kafka-connector task:

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

Verify the result.

• Query data in the MaxCompute table.

  Run the following commands on the MaxCompute client (odpscmd) or another tool that can run MaxCompute SQL to query the data write result:

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

  The following result is returned:

  # The last two records are displayed. This is because the skip_error parameter is set to true. The data with the id of 101 is not written to the MaxCompute table, and subsequent records are not blocked from being written to the MaxCompute table. 
  +-------+------------+------------+------------+------+------------+----+
  | 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 |
  +-------+------------+------------+------------+------+------------+----+

• Query messages in the runtime_error topic.

  In the $KAFKA_HOME/bin/ directory, run the following command to view the message write result:

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

  The following result is returned:

  # Abnormal data is written to the runtime_error topic.
  {"id":101,"name":"json-4","extendinfos":"null"}