ApsaraDB for SelectDB:Gunakan Routine Load untuk mengimpor data

Perilaku mode ketat

Mode ketat mengatur cara penanganan kegagalan konversi tipe.

Mengimpor kolom TINYINT (NULL diizinkan):

Mengimpor kolom DECIMAL(1,0) (NULL diizinkan):

Nilai 10 melebihi rentang DECIMAL(1,0) tetapi tidak difilter oleh mode ketat, karena nilainya sendiri lolos konversi tipe. Nilai tersebut difilter kemudian selama proses ekstrak, transformasi, dan muat (ETL).

Muat data CSV

1. Buat tabel tujuan:

   CREATE TABLE test_table (
       id      INT,
       name    VARCHAR(50),
       age     INT,
       address VARCHAR(50),
       url     VARCHAR(500)
   )
   UNIQUE KEY(`id`, `name`)
   DISTRIBUTED BY HASH(id) BUCKETS 4
   PROPERTIES ("replication_num" = "1");

2. Buat pekerjaan Routine Load yang membaca dari awal semua partisi:

   CREATE ROUTINE LOAD example_db.test1 ON test_table
   COLUMNS TERMINATED BY ",",
   COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100)
   PROPERTIES (
       "desired_concurrent_number" = "3",
       "max_batch_interval"        = "20",
       "max_batch_rows"            = "300000",
       "max_batch_size"            = "209715200",
       "strict_mode"               = "false"
   )
   FROM KAFKA (
       "kafka_broker_list"              = "broker1:9092,broker2:9092,broker3:9092",
       "kafka_topic"                    = "my_topic",
       "property.kafka_default_offsets" = "OFFSET_BEGINNING"
   );

3. Buat pekerjaan Routine Load dengan mode ketat diaktifkan:

   CREATE ROUTINE LOAD example_db.test2 ON test_table
   COLUMNS TERMINATED BY ",",
   COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100)
   PROPERTIES (
       "desired_concurrent_number" = "3",
       "max_batch_interval"        = "20",
       "max_batch_rows"            = "300000",
       "max_batch_size"            = "209715200",
       "strict_mode"               = "true"
   )
   FROM KAFKA (
       "kafka_broker_list"              = "broker1:9092,broker2:9092,broker3:9092",
       "kafka_topic"                    = "my_topic",
       "property.kafka_default_offsets" = "OFFSET_BEGINNING"
   );

4. Mulai mengonsumsi dari timestamp tertentu:

   CREATE ROUTINE LOAD example_db.test4 ON test_table
   PROPERTIES (
       "desired_concurrent_number" = "3",
       "max_batch_interval"        = "30",
       "max_batch_rows"            = "300000",
       "max_batch_size"            = "209715200"
   )
   FROM KAFKA (
       "kafka_broker_list"             = "broker1:9092,broker2:9092",
       "kafka_topic"                   = "my_topic",
       "property.kafka_default_offset" = "2024-01-21 10:00:00"
   );

Muat data JSON

Routine Load mendukung dua struktur pesan JSON:

• Objek JSON tunggal — satu catatan per pesan:

  {"key1":"value1","key2":"value2","key3":"value3"}

• Array JSON — beberapa catatan per pesan:

  [
      {"key1":"value11","key2":"value12","key3":"value13","key4":14},
      {"key1":"value21","key2":"value22","key3":"value23","key4":24}
  ]

Untuk mode impor multi-tabel, tambahkan prefiks nama tabel tujuan pada setiap pesan:

table_name|{"key1":"value1","key2":"value2","key3":"value3"}

Contoh: muat data JSON

1. Buat tabel tujuan:

   CREATE TABLE `example_tbl` (
       `category`  VARCHAR(24)  NULL,
       `author`    VARCHAR(24)  NULL,
       `timestamp` BIGINT(20)   NULL,
       `dt`        INT(11)      NULL,
       `price`     DOUBLE REPLACE
   ) ENGINE=OLAP
   AGGREGATE KEY(`category`, `author`, `timestamp`, `dt`)
   PARTITION BY RANGE(`dt`) (
       PARTITION p0        VALUES [("-2147483648"), ("20230509")),
       PARTITION p20200509 VALUES [("20230509"), ("20231010")),
       PARTITION p20200510 VALUES [("20231010"), ("20231211")),
       PARTITION p20200511 VALUES [("20231211"), ("20240512"))
   )
   DISTRIBUTED BY HASH(`category`, `author`, `timestamp`) BUCKETS 4;

2. Publikasikan kedua jenis pesan ke topik Kafka:

   {"category":"value1331","author":"value1233","timestamp":1700346050,"price":1413}

   [
       {"category":"value13z2","author":"vaelue13","timestamp":1705645251,"price":14330},
       {"category":"lvalue211","author":"lvalue122","timestamp":1684448450,"price":24440}
   ]

3. Muat dalam mode sederhana (nama bidang sesuai dengan nama kolom):

   CREATE ROUTINE LOAD example_db.test_json_label_1 ON example_tbl
   COLUMNS(category, price, author)
   PROPERTIES (
       "desired_concurrent_number" = "3",
       "max_batch_interval"        = "20",
       "max_batch_rows"            = "300000",
       "max_batch_size"            = "209715200",
       "strict_mode"               = "false",
       "format"                    = "json"
   )
   FROM KAFKA (
       "kafka_broker_list"  = "broker1:9092,broker2:9092,broker3:9092",
       "kafka_topic"        = "my_topic",
       "kafka_partitions"   = "0,1,2",
       "kafka_offsets"      = "0,0,0"
   );

4. Muat dengan ekstraksi jalur JSON eksplisit dan pengupasan array luar:

   Kolom dt tidak muncul dalam data sumber. Nilainya diturunkan dari timestamp menggunakan dt=from_unixtime(timestamp,'%Y%m%d') dalam klausa COLUMNS.

   CREATE ROUTINE LOAD example_db.test_json_label_3 ON example_tbl
   COLUMNS(category, author, price, timestamp, dt=from_unixtime(timestamp, '%Y%m%d'))
   PROPERTIES (
       "desired_concurrent_number" = "3",
       "max_batch_interval"        = "20",
       "max_batch_rows"            = "300000",
       "max_batch_size"            = "209715200",
       "strict_mode"               = "false",
       "format"                    = "json",
       "jsonpaths"                 = "[\"$.category\",\"$.author\",\"$.price\",\"$.timestamp\"]",
       "strip_outer_array"         = "true"
   )
   FROM KAFKA (
       "kafka_broker_list"  = "broker1:9092,broker2:9092,broker3:9092",
       "kafka_topic"        = "my_topic",
       "kafka_partitions"   = "0,1,2",
       "kafka_offsets"      = "0,0,0"
   );

Koneksi ke kluster Kafka dengan otentikasi

ApsaraDB for SelectDB menggunakan library klien C++ librdkafka untuk terhubung ke Kafka. Untuk properti konfigurasi yang didukung, lihat referensi konfigurasi librdkafka.

Otentikasi SSL

Unggah file sertifikat yang diperlukan terlebih dahulu, lalu buat pekerjaan Routine Load.

1. Unggah file sertifikat:

   CREATE FILE "ca.pem"     PROPERTIES ("url" = "https://example_url/kafka-key/ca.pem",     "catalog" = "kafka");
   CREATE FILE "client.key" PROPERTIES ("url" = "https://example_url/kafka-key/client.key", "catalog" = "kafka");
   CREATE FILE "client.pem" PROPERTIES ("url" = "https://example_url/kafka-key/client.pem", "catalog" = "kafka");

2. Buat pekerjaan Routine Load:

   Properti	Wajib	Deskripsi
   property.security.protocol	Selalu	Atur ke ssl.
   property.ssl.ca.location	Selalu	Jalur ke sertifikat CA yang mengotentikasi kunci publik broker Kafka.
   property.ssl.certificate.location	Hanya jika otentikasi klien diaktifkan di server Kafka	Jalur ke sertifikat kunci publik klien.
   property.ssl.key.location	Hanya jika otentikasi klien diaktifkan di server Kafka	Jalur ke file kunci privat klien.
   property.ssl.key.password	Hanya jika otentikasi klien diaktifkan di server Kafka	Kata sandi untuk kunci privat klien.

   CREATE ROUTINE LOAD db1.job1 ON tbl1
   PROPERTIES (
       "desired_concurrent_number" = "1"
   )
   FROM KAFKA (
       "kafka_broker_list"                 = "broker1:9091,broker2:9091",
       "kafka_topic"                       = "my_topic",
       "property.security.protocol"        = "ssl",
       "property.ssl.ca.location"          = "FILE:ca.pem",
       "property.ssl.certificate.location" = "FILE:client.pem",
       "property.ssl.key.location"         = "FILE:client.key",
       "property.ssl.key.password"         = "abcdefg"
   );

Otentikasi PLAIN

CREATE ROUTINE LOAD db1.job1 ON tbl1
PROPERTIES (
    "desired_concurrent_number" = "1"
)
FROM KAFKA (
    "kafka_broker_list"          = "broker1:9092,broker2:9092",
    "kafka_topic"                = "my_topic",
    "property.security.protocol" = "SASL_PLAINTEXT",
    "property.sasl.mechanism"    = "PLAIN",
    "property.sasl.username"     = "admin",
    "property.sasl.password"     = "admin"
);

Otentikasi Kerberos

Sebelum membuat pekerjaan, deploy klien Kerberos kinit di semua node dalam kluster ApsaraDB for SelectDB Anda, konfigurasikan krb5.conf, dan tentukan layanan Key Distribution Center (KDC).

CREATE ROUTINE LOAD db1.job1 ON tbl1
PROPERTIES (
    "desired_concurrent_number" = "1"
)
FROM KAFKA (
    "kafka_broker_list"                   = "broker1:9092,broker2:9092",
    "kafka_topic"                         = "my_topic",
    "property.security.protocol"          = "SASL_PLAINTEXT",
    "property.sasl.kerberos.service.name" = "kafka",
    "property.sasl.kerberos.keytab"       = "/etc/krb5.keytab",
    "property.sasl.kerberos.principal"    = "id@your.com"
);