MaxCompute:,

Persiapkan data.

• Gunakan client lokal (odpscmd) atau alat lain yang dapat menjalankan perintah SQL MaxCompute untuk terhubung ke MaxCompute dan membuat tabel target.

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

• Buat data Kafka.

  Di direktori $KAFKA_HOME/bin/, jalankan perintah berikut untuk membuat topik Kafka. Contoh ini menggunakan topic_text sebagai nama topik.

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

  Jalankan perintah berikut untuk membuat pesan Kafka.

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

(Opsional) Jalankan layanan Kafka-connector. Untuk informasi selengkapnya, lihat Konfigurasikan dan jalankan layanan konektor Kafka.

Buat dan konfigurasikan file odps-sink-connector.json. Kemudian, unggah file odps-sink-connector.json ke lokasi apa pun, misalnya ke path $KAFKA_HOME/config.

Kode berikut memberikan contoh file odps-sink-connector.json. Untuk informasi selengkapnya tentang file odps-sink-connector.json, lihat Konfigurasikan dan jalankan task konektor Kafka.

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

Jalankan perintah berikut untuk memulai task migrasi data konektor Kafka.

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

Verifikasi hasil.

Gunakan client lokal (odpscmd) atau alat lain yang dapat menjalankan perintah SQL MaxCompute untuk terhubung ke MaxCompute, lalu jalankan perintah berikut untuk mengkueri data dan memverifikasi hasil.

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

Output berikut dikembalikan:

# Karena parameter mode dalam file konfigurasi odps-sink-connector.json diatur ke VALUE, hanya konten value yang disimpan. Bidang key bernilai 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 |
+-------+------------+------------+-----+-------+----+

Persiapkan data.

• Gunakan client lokal (odpscmd) atau alat lain yang dapat menjalankan perintah SQL MaxCompute untuk terhubung ke MaxCompute dan membuat tabel target.

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

• Tulis data ke Kafka.

  Di direktori $KAFKA_HOME/bin/, jalankan perintah berikut untuk membuat topik Kafka bernama topic_csv.

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

  Jalankan perintah berikut untuk membuat pesan 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

(Opsional) Jalankan layanan Kafka-connector. Untuk informasi selengkapnya, lihat Konfigurasikan dan jalankan layanan konektor Kafka.

Buat dan konfigurasikan file odps-sink-connector.json, lalu unggah file odps-sink-connector.json ke lokasi apa pun. Topik ini menggunakan path $KAFKA_HOME/config sebagai contoh.

Kode berikut memberikan contoh file odps-sink-connector.json. Untuk informasi selengkapnya tentang file odps-sink-connector.json, lihat Konfigurasikan dan jalankan task konektor Kafka.

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

Jalankan perintah berikut untuk memulai task migrasi data konektor Kafka.

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

Verifikasi hasil.

Gunakan client lokal (odpscmd) atau alat lain yang dapat menjalankan perintah SQL MaxCompute untuk terhubung ke MaxCompute, lalu jalankan perintah berikut untuk mengkueri data dan memverifikasi hasil.

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

Output berikut dikembalikan:

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

Persiapkan data.

• Gunakan client lokal (odpscmd) atau alat lain yang dapat menjalankan perintah SQL MaxCompute untuk terhubung ke MaxCompute dan membuat tabel target.

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

• Buat data Kafka.

  Di direktori $KAFKA_HOME/bin/, jalankan perintah berikut untuk membuat topik Kafka. Contoh ini menggunakan topic_json sebagai nama topik.

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

  Jalankan perintah berikut untuk membuat pesan 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"}

(Opsional) Jalankan layanan Kafka-connector. Untuk informasi selengkapnya, lihat Konfigurasikan dan jalankan layanan konektor Kafka.

Buat dan konfigurasikan file odps-sink-connector.json. Kemudian, unggah file odps-sink-connector.json ke lokasi apa pun, misalnya ke path $KAFKA_HOME/config.

Kode berikut memberikan contoh file odps-sink-connector.json. Untuk informasi selengkapnya tentang file odps-sink-connector.json, lihat Konfigurasikan dan jalankan task konektor Kafka.

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

Jalankan perintah berikut untuk memulai task migrasi data konektor Kafka.

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

Verifikasi hasil.

Gunakan client lokal (odpscmd) atau alat lain yang dapat menjalankan perintah SQL MaxCompute untuk terhubung ke MaxCompute, lalu jalankan perintah berikut untuk mengkueri data dan memverifikasi hasil.

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

Output berikut dikembalikan:

# Data JSON berhasil ditulis ke bidang 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 |
+-------+------------+------------+-----+-------+----+

Persiapkan data.

• Gunakan client lokal (odpscmd) atau alat lain yang dapat menjalankan perintah SQL MaxCompute untuk terhubung ke MaxCompute dan membuat tabel target.

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

• Buat data Kafka.

  Di direktori $KAFKA_HOME/bin/, jalankan perintah berikut untuk membuat topik Kafka. Contoh ini menggunakan topic_flatten sebagai nama topik.

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

  Jalankan perintah berikut untuk membuat pesan 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"}}

(Opsional) Jalankan layanan Kafka-connector. Untuk informasi selengkapnya, lihat Konfigurasikan dan jalankan layanan konektor Kafka.

Buat dan konfigurasikan file odps-sink-connector.json, lalu unggah file odps-sink-connector.json ke lokasi apa pun. Topik ini menggunakan path $KAFKA_HOME/config sebagai contoh.

Kode berikut memberikan contoh file odps-sink-connector.json. Untuk informasi selengkapnya tentang file odps-sink-connector.json, lihat Konfigurasikan dan jalankan task konektor Kafka.

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

Jalankan perintah berikut untuk memulai task konektor Kafka.

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

Verifikasi hasil.

Gunakan client lokal (odpscmd) atau alat lain yang dapat menjalankan perintah SQL MaxCompute untuk terhubung ke MaxCompute, lalu jalankan perintah berikut untuk mengkueri data dan memverifikasi hasil.

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

Berikut ini menunjukkan hasilnya:

# Data JSON diurai dan ditulis ke tabel MaxCompute, dengan extendinfo sebagai bidang JSON yang mendukung 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 |
+-------+------------+--------+-----+------+------------+----+

Persiapkan data.

• Gunakan client lokal (odpscmd) atau alat lain yang dapat menjalankan perintah SQL MaxCompute untuk terhubung ke MaxCompute dan membuat tabel target.

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

• Buat data Kafka.

  Di direktori $KAFKA_HOME/bin/, jalankan perintah berikut untuk membuat topik Kafka.

  • Topik topic_abnormal.

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

  • Topik pesan untuk exception runtime_error.

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

    Catatan

    Jika terjadi error selama operasi penulisan data, data abnormal ditulis ke topik runtime_error. Jenis error ini biasanya disebabkan oleh ketidaksesuaian antara data Kafka dan skema tabel MaxCompute.

  Jalankan perintah berikut untuk membuat pesan Kafka.

  Salah satu pesan dalam perintah berikut tidak sesuai dengan skema tabel MaxCompute target.

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

(Opsional) Jalankan layanan Kafka-connector. Untuk informasi selengkapnya, lihat Konfigurasikan dan jalankan layanan konektor Kafka.

Buat dan konfigurasikan file odps-sink-connector.json, lalu unggah file odps-sink-connector.json ke lokasi apa pun. Topik ini menggunakan path $KAFKA_HOME/config sebagai contoh.

Kode berikut memberikan contoh file odps-sink-connector.json. Untuk informasi selengkapnya tentang file odps-sink-connector.json, lihat Konfigurasikan dan jalankan task konektor Kafka.

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

Jalankan perintah berikut untuk memulai task konektor Kafka.

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

Verifikasi hasil.

• Kueri data tabel MaxCompute

  Gunakan client lokal (odpscmd) atau alat lain yang dapat menjalankan perintah SQL MaxCompute untuk terhubung ke MaxCompute, lalu jalankan perintah berikut untuk mengkueri data dan memverifikasi hasil.

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

  Output berikut dikembalikan:

  # Seperti yang terlihat dari hasil, data dengan ID 101 tidak ditulis ke MaxCompute karena tidak sesuai dengan skema tabel.
  # Karena parameter runtime.error.topic.name dikonfigurasi, proses tidak terblokir, dan data berikutnya ditulis berhasil.
  +-------+------------+------------+------------+------+------------+----+
  | 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 |
  +-------+------------+------------+------------+------+------------+----+

• Kueri pesan di topik runtime_error

  Di direktori $KAFKA_HOME/bin/, jalankan perintah berikut untuk melihat pesan.

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

  Hasil berikut dikembalikan:

  # Data abnormal berhasil ditulis ke antrian pesan runtime_error.
  {"id":101,"name":"json-4","extendinfos":"null"}