,

Prepare the data.

• Use a local client (odpscmd) or another tool that can run MaxCompute SQL commands to connect to MaxCompute and create the target table.

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

• Create the Kafka data.

  In the $KAFKA_HOME/bin/ directory, run the following command to create a Kafka topic. This example uses topic_text as the topic name.

  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. Then, upload the odps-sink-connector.json file to any location, such as the $KAFKA_HOME/config path.

The following code provides an example of the odps-sink-connector.json file. For more information about the odps-sink-connector.json file, see Configure and start the 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": "<yourAccessKeyId>",
      "access_key": "<yourAccessKeySecret>",
      "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 data migration 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.

Use a local client (odpscmd) or another tool that can run MaxCompute SQL commands to connect to MaxCompute, and then run the following command to query the data and verify the result.

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

The following output is returned:

# Because the mode parameter in the odps-sink-connector.json configuration file is set to VALUE, only the content of the value is retained. 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 the data.

• Use a local client (odpscmd) or another tool that can run MaxCompute SQL commands to connect to MaxCompute and create the target table.

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

• Write data to Kafka.

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

  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 then upload the odps-sink-connector.json file to any location. This topic uses the $KAFKA_HOME/config path as an example.

The following code provides an example of the odps-sink-connector.json file. For more information about the odps-sink-connector.json file, see Configure and start the 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": "<yourAccessKeyId>",
      "access_key": "<yourAccessKeySecret>",
      "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 data migration 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.

Use a local client (odpscmd) or another tool that can run MaxCompute SQL commands to connect to MaxCompute, and then run the following command to query the data and verify the result.

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

The following output 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 the data.

• Use a local client (odpscmd) or another tool that can run MaxCompute SQL commands to connect to MaxCompute and create the target table.

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

• Create the Kafka data.

  In the $KAFKA_HOME/bin/ directory, run the following command to create a Kafka topic. This example uses topic_json as the topic name.

  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. Then, upload the odps-sink-connector.json file to any location, such as the $KAFKA_HOME/config path.

The following code provides an example of the odps-sink-connector.json file. For more information about the odps-sink-connector.json file, see Configure and start the 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": "<yourAccessKeyId>",
      "access_key": "<yourAccessKeySecret>",
      "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 data migration 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.

Use a local client (odpscmd) or another tool that can run MaxCompute SQL commands to connect to MaxCompute, and then run the following command to query the data and verify the result.

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

The following output is returned:

# The JSON data is successfully written 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 the data.

• Use a local client (odpscmd) or another tool that can run MaxCompute SQL commands to connect to MaxCompute and create the target table.

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

• Create the Kafka data.

  In the $KAFKA_HOME/bin/ directory, run the following command to create a Kafka topic. This example uses topic_flatten as the topic name.

  ./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 then upload the odps-sink-connector.json file to any location. This topic uses the $KAFKA_HOME/config path as an example.

The following code provides an example of the odps-sink-connector.json file. For more information about the odps-sink-connector.json file, see Configure and start the 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": "<yourAccessKeyId>",
      "access_key": "<yourAccessKeySecret>",
      "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.

Use a local client (odpscmd) or another tool that can run MaxCompute SQL commands to connect to MaxCompute, and then run the following command to query the data and verify the result.

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

The following shows the result:

# The JSON data is parsed and written to a MaxCompute table, with extendinfo as a JSON field that supports nesting.
+-------+------------+--------+-----+------+------------+----+
| 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 the data.

• Use a local client (odpscmd) or another tool that can run MaxCompute SQL commands to connect to MaxCompute and create the target table.

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

• Create the Kafka data.

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

  • The topic_abnormal topic.

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

  • Message topic for runtime_error exceptions.

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

    Note

    If an error occurs during a data write operation, the abnormal data is written to the runtime_error topic. This type of error is usually caused by a mismatch between the Kafka data and the MaxCompute table schema.

  Run the following command to create Kafka messages.

  One of the messages in the following command does not match the schema of the target 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 then upload the odps-sink-connector.json file to any location. This topic uses the $KAFKA_HOME/config path as an example.

The following code provides an example of the odps-sink-connector.json file. For more information about the odps-sink-connector.json file, see Configure and start the 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": "<yourAccessKeyId>",
    "access_key": "<yourAccessKeySecret>",
    "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 the MaxCompute table data

  Use a local client (odpscmd) or another tool that can run MaxCompute SQL commands to connect to MaxCompute, and then run the following command to query the data and verify the result.

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

  The following output is returned:

  # As you can see from the results, the data with ID 101 was not written to MaxCompute because it did not match the table schema.
  # Because the runtime.error.topic.name parameter was configured, the process was not blocked, and subsequent data was written successfully.
  +-------+------------+------------+------------+------+------------+----+
  | 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 messages.

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

  The following result is returned:

  # The abnormal data is successfully written to the runtime_error message queue.
  {"id":101,"name":"json-4","extendinfos":"null"}