MaxCompute:,

Write TEXT data

1. Create the target MaxCompute table. Run the following SQL in odpscmd or any MaxCompute SQL client:

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

2. Create a Kafka topic and send test messages. Run the following commands in the $KAFKA_HOME/bin/ directory:

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

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

3. (Optional) Start the connector service if it is not already running. See Set up and start the connector service.

4. Create odps-sink-connector.json and save it to $KAFKA_HOME/config/:

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

5. Start the sink 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

6. Verify the result. Run the following SQL in odpscmd or any MaxCompute SQL client:

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

   Expected output — because mode is VALUE, the key column 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 |
   +-------+------------+------------+-----+-------+------------------+

Write CSV data

1. 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);

2. Create a Kafka topic and send test messages:

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

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

3. (Optional) Start the connector service if not already running.

4. Create odps-sink-connector.json:

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

5. Start the sink 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

6. Verify the result:

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

   Expected output:

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

Write JSON data

1. Create the target table. The value column uses the MaxCompute JSON type to store structured data:

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

2. Create a Kafka topic and send test messages:

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

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

3. (Optional) Start the connector service if not already running.

4. Create odps-sink-connector.json:

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

5. Start the sink 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

6. Verify the result:

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

   Expected output — the JSON payload is stored in the value column:

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

Write FLATTEN data

FLATTEN mode parses each JSON message and maps its top-level keys to matching columns in the MaxCompute table. Nested objects are stored as JSON columns.

1. Create the target table. Custom columns (id, name, extendinfo) match the JSON keys in the Kafka messages:

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

2. Create a Kafka topic and send test messages:

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

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

3. (Optional) Start the connector service if not already running.

4. Create odps-sink-connector.json:

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

5. Start the sink 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

6. Verify the result:

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

   Expected output — the JSON payload is flattened into columns, with extendinfo stored as a nested 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 |
   +---------------+-----------+--------+-----+--------+--------------------------------+------------------+

Handle abnormal data

This example demonstrates how to route schema-mismatched records to a dedicated error topic instead of blocking the pipeline.

1. Create the target table and two Kafka topics — one for normal data and one for errors:

   Records that fail to write are redirected to runtime_error. This usually happens when the Kafka message structure does not match the MaxCompute table schema.

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

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

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

2. Send test messages. The second message uses extendinfos (note the typo) instead of extendinfo, which does not match the table schema:

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

3. (Optional) Start the connector service if not already running.

4. Create odps-sink-connector.json with runtime.error.topic.name set to redirect failed records:

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

5. Start the sink 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

6. Verify the result. Query the MaxCompute table. The record with ID 101 is absent because it failed schema validation, but processing continued for records 100 and 102:

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

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

   Inspect the failed record in the runtime_error topic:

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

   {"id":101,"name":"json-4","extendinfos":"null"}

   The mismatched record is written to the error topic, allowing you to investigate and reprocess it.