ApsaraDB for SelectDB:Routine Load を使用したデータのインポート

Strict モードの動作

Strict モードは、型変換失敗時の処理方法を制御します。

TINYINT カラム（NULL 許容）へのインポート：

DECIMAL(1,0) カラム（NULL 許容）へのインポート：

値 10 は DECIMAL(1,0) の範囲を超えていますが、Strict モードではフィルターされません。これは、値自体が型変換を通過するためです。この値は、抽出・変換・書き出し（ETL）プロセス中に後からフィルターされます。

CSV データのロード

1. 送信先テーブルを作成します。

   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. すべてのパーティションの先頭から読み取る Routine Load ジョブを作成します。

   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. Strict モードを有効にして Routine Load ジョブを作成します。

   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. 特定のタイムスタンプから消費を開始します。

   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"
   );

JSON データのロード

Routine Load は次の 2 種類の JSON メッセージ構造をサポートしています。

• 単一の JSON オブジェクト — メッセージごとに 1 レコード：

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

• JSON 配列 — メッセージごとに複数レコード：

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

複数テーブルインポートモードでは、各メッセージの先頭に送信先テーブル名を付加します。

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

例：JSON データのロード

1. 送信先テーブルを作成します。

   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. 両方のメッセージタイプを 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. シンプルモードでロードします（フィールド名がカラム名と一致）。

   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. 明示的な JSON パス抽出と外側の配列のストリップを伴ってロードします。

   dt カラムはソースデータに存在しません。その値は、COLUMNS 句内の timestamp を使用して dt=from_unixtime(timestamp,'%Y%m%d') で導出されます。

   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"
   );

認証付き Kafka クラスターへの接続

ApsaraDB for SelectDB は Kafka への接続に librdkafka C++ クライアントライブラリを使用します。サポートされている構成プロパティについては、「librdkafka 構成リファレンス」をご参照ください。

SSL 認証

必要な証明書ファイルを最初にアップロードしてから、Routine Load ジョブを作成します。

1. 証明書ファイルをアップロードします。

   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. Routine Load ジョブを作成します。

   プロパティ	必須	説明
   property.security.protocol	常に	ssl に設定します。
   property.ssl.ca.location	常に	Kafka ブローカーの公開鍵を認証する CA 証明書のパス。
   property.ssl.certificate.location	Kafka サーバーでクライアント認証が有効になっている場合のみ	クライアント公開鍵証明書のパス。
   property.ssl.key.location	Kafka サーバーでクライアント認証が有効になっている場合のみ	クライアント秘密鍵ファイルのパス。
   property.ssl.key.password	Kafka サーバーでクライアント認証が有効になっている場合のみ	クライアント秘密鍵のパスワード。

   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"
   );

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"
);

Kerberos 認証

ジョブを作成する前に、ApsaraDB for SelectDB クラスターのすべてのノードに Kerberos クライアント kinit をデプロイし、krb5.conf を構成して、キー配布センター（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"
);