MaxCompute：Impor data Kafka ke MaxCompute dalam mode offline atau real-time

Persiapkan Data.

• Buat tabel MaxCompute menggunakan Klien MaxCompute (odpscmd) atau alat lain yang dapat menjalankan SQL MaxCompute.

  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 bernama topic_text.

  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) Mulai layanan Kafka-connector. Untuk informasi lebih lanjut, lihat Konfigurasikan dan Mulai Layanan Kafka-connector.

Buat dan konfigurasikan file odps-sink-connector.json, lalu unggah file tersebut ke lokasi mana pun. Dalam contoh ini, file odps-sink-connector.json diunggah ke direktori $KAFKA_HOME/config.

Kode berikut menampilkan isi file odps-sink-connector.json. Untuk detail lebih lanjut tentang file odps-sink-connector.json, lihat Konfigurasikan dan mulai tugas 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": "<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 tugas 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

Verifikasi Hasilnya.

Jalankan perintah berikut pada Klien MaxCompute (odpscmd) atau alat lain yang dapat menjalankan SQL MaxCompute untuk memeriksa hasil penulisan data:

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

Hasil berikut dikembalikan:

# Nilai mode dalam file konfigurasi odps-sink-connector.json adalah VALUE. Oleh karena itu, hanya nilai yang disimpan dan bidang kunci adalah 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.

• Buat tabel MaxCompute tujuan menggunakan Klien MaxCompute (odpscmd) atau alat lain yang dapat menjalankan 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);

• Buat data 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) Mulai layanan Kafka-connector. Untuk informasi lebih lanjut, lihat Konfigurasikan dan Mulai Layanan Kafka-connector.

Buat dan konfigurasikan file odps-sink-connector.json, lalu unggah file tersebut ke lokasi mana pun. Dalam contoh ini, file odps-sink-connector.json diunggah ke direktori $KAFKA_HOME/config.

Kode berikut menampilkan isi file odps-sink-connector.json. Untuk detail lebih lanjut tentang file odps-sink-connector.json, lihat Konfigurasikan dan mulai tugas 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": "<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 tugas 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

Verifikasi Hasilnya.

Jalankan perintah berikut pada Klien MaxCompute (odpscmd) atau alat lain yang dapat menjalankan SQL MaxCompute untuk memeriksa hasil penulisan data:

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

Hasil 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.

• Buat tabel MaxCompute tujuan menggunakan Klien MaxCompute (odpscmd) atau alat lain yang dapat menjalankan SQL MaxCompute.

  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 bernama topic_json.

  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) Mulai layanan Kafka-connector. Untuk informasi lebih lanjut, lihat Konfigurasikan dan Mulai Layanan Kafka-connector.

Buat dan konfigurasikan file odps-sink-connector.json, lalu unggah file tersebut ke lokasi mana pun. Dalam contoh ini, file odps-sink-connector.json diunggah ke direktori $KAFKA_HOME/config.

Kode berikut menampilkan isi file odps-sink-connector.json. Untuk detail lebih lanjut tentang file odps-sink-connector.json, lihat Konfigurasikan dan mulai tugas 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": "<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 tugas 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

Verifikasi Hasilnya.

Jalankan perintah berikut pada Klien MaxCompute (odpscmd) atau alat lain yang dapat menjalankan SQL MaxCompute untuk memeriksa hasil penulisan data:

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

Hasil berikut dikembalikan:

# Tulis data JSON 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.

• Buat tabel MaxCompute tujuan menggunakan Klien MaxCompute (odpscmd) atau alat lain yang dapat menjalankan 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);

• Buat data Kafka.

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

  ./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) Mulai layanan Kafka-connector. Untuk informasi lebih lanjut, lihat Konfigurasikan dan Mulai Layanan Kafka-connector.

Buat dan konfigurasikan file odps-sink-connector.json, lalu unggah file tersebut ke lokasi mana pun. Dalam contoh ini, file odps-sink-connector.json diunggah ke direktori $KAFKA_HOME/config.

Kode berikut menampilkan isi file odps-sink-connector.json. Untuk detail lebih lanjut tentang file odps-sink-connector.json, lihat Konfigurasikan dan mulai tugas 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": "<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 tugas 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

Verifikasi Hasilnya.

Jalankan perintah berikut pada Klien MaxCompute (odpscmd) atau alat lain yang dapat menjalankan SQL MaxCompute untuk memeriksa hasil penulisan data:

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

Hasil berikut dikembalikan:

# Data JSON diurai dan ditulis ke tabel MaxCompute. Bidang exteninfo dalam format JSON bertingkat dapat berupa bidang 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 |
+-------+------------+--------+-----+------+------------+----+

Persiapkan Data.

• Buat tabel MaxCompute tujuan menggunakan Klien MaxCompute (odpscmd) atau alat lain yang dapat menjalankan 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);

• Buat data Kafka.

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

  • 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

    Catatan

    Jika terjadi kesalahan yang tidak diketahui saat menulis data, data abnormal akan ditulis ke topik runtime_error. Pada umumnya, kesalahan yang tidak diketahui terjadi karena format data Kafka tidak sesuai dengan format tabel MaxCompute.

  Jalankan perintah berikut untuk membuat pesan Kafka:

  Dalam pesan berikut, format data satu pesan tidak sama dengan format tabel 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"}} 

(Opsional) Mulai layanan Kafka-connector. Untuk informasi lebih lanjut, lihat Konfigurasikan dan Mulai Layanan Kafka-connector.

Buat dan konfigurasikan file odps-sink-connector.json, lalu unggah file tersebut ke lokasi mana pun. Dalam contoh ini, file odps-sink-connector.json diunggah ke direktori $KAFKA_HOME/config.

Kode berikut menampilkan isi file odps-sink-connector.json. Untuk detail lebih lanjut tentang file odps-sink-connector.json, lihat Konfigurasikan dan mulai tugas 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": "<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 tugas 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

Verifikasi Hasilnya.

• Kueri Data dalam Tabel MaxCompute.

  Jalankan perintah berikut pada Klien MaxCompute (odpscmd) atau alat lain yang dapat menjalankan SQL MaxCompute untuk memeriksa hasil penulisan data:

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

  Hasil berikut dikembalikan:

  # Dua rekaman terakhir ditampilkan. Hal ini disebabkan parameter skip_error diatur ke true. Data dengan id 101 tidak ditulis ke tabel MaxCompute, dan rekaman selanjutnya tidak diblokir dari penulisan ke tabel MaxCompute.
  +-------+------------+------------+------------+------+------------+----+
  | 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 dalam Topik runtime_error.

  Di direktori $KAFKA_HOME/bin/, jalankan perintah berikut untuk melihat hasil penulisan pesan:

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

  Hasil berikut dikembalikan:

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