ApsaraDB for SelectDB:ルーチンロードを使用したデータのインポート

ルーチンロードジョブの作成

ルーチンロードを使用するには、ルーチンロードジョブを作成する必要があります。ルーチンロードジョブは、ルーチン スケジューリングに基づいてタスクを継続的にスケジュールします。各タスクは、特定の数の Kafka メッセージを消費します。

構文

CREATE ROUTINE LOAD [db.]job_name ON tbl_name
[merge_type]
[load_properties]
[job_properties]
FROM data_source [data_source_properties]

パラメータ

[column_separator],
[columns_mapping],
[preceding_filter],
[where_predicates],
[partitions],
[DELETE ON],
[ORDER BY]

PROPERTIES (
    "key1" = "val1",
    "key2" = "val2"
)

FROM KAFKA
(
    "key1" = "val1",
    "key2" = "val2"
)

"kafka_partitions" = "0,1,2,3",
"kafka_offsets" = "101,0,OFFSET_BEGINNING,OFFSET_END"

"kafka_partitions" = "0,1,2,3",
"kafka_offsets" = "2021-05-22 11:00:00,2021-05-22 11:00:00,2021-05-22 11:00:00"

"property.client.id" = "12345",
"property.ssl.ca.location" = "FILE:ca.pem"

例

ルーチンロードジョブの作成

1. ApsaraDB for SelectDBインスタンスにデータを読み込むテーブルを作成します。サンプルコード:

   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. 次のサンプルコードのパラメーターを構成して、データを読み込みます。

   • example_dbデータベースのtest_tableテーブルに対して、test1という名前のRoutine Loadジョブを作成します。グループID、クライアントID、および列区切り文字を指定し、デフォルトですべてのパーティションのデータを自動的に使用するようにシステムを有効にし、データが使用可能な開始オフセットからパーティションをサブスクライブします。サンプルコード:

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

   • example_dbデータベースのtest_tableテーブルに対して、test2という名前の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"
     );

   • 指定した時点からパーティションのデータを使用します。サンプルコード:

     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つだけ含まれており、JSONオブジェクトです。

  単一テーブルインポートモードを使用する場合、JSONデータは次の形式になります。このモードでは、ON TABLE_NAMEステートメントを実行することでテーブル名が指定されます。

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

  動的インポートモードまたは複数テーブルインポートモードを使用する場合、JSONデータは次の形式になります。このモードでは、テーブル名は指定されません。

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

• JSONデータは、複数のデータレコードを含む配列です。

  単一テーブルインポートモードを使用する場合、JSONデータは次の形式になります。このモードでは、ON TABLE_NAMEステートメントを実行することでテーブル名が指定されます。

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

  動的インポートモードまたは複数テーブルインポートモードを使用する場合、JSONデータは次の形式になります。このモードでは、テーブル名は指定されません。

     table_name|[
      {   
          "key1":"value11",
          "key2":"value12",
          "key3":"value13",
          "key4":14
      },
      {
          "key1":"value21",
          "key2":"value22",
          "key3":"value23",
          "key4":24
      },
      {
          "key1":"value31",
          "key2":"value32",
          "key3":"value33",
          "key4":34
      }
  ]

JSON データをインポートします。

1. ApsaraDB for SelectDBインスタンスにデータを読み込みたいテーブルを作成します。サンプルコード:

   CREATE TABLE `example_tbl` (
      `category` varchar(24) NULL COMMENT "",
      `author` varchar(24) NULL COMMENT "",
      `timestamp` bigint(20) NULL COMMENT "",
      `dt` int(11) NULL COMMENT "",
      `price` double REPLACE
   ) ENGINE=OLAP
   AGGREGATE KEY(`category`,`author`,`timestamp`,`dt`)
   COMMENT "OLAP"  // OLAP
   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. 上記の2種類のJSONデータをトピックにインポートします。サンプルコード:

   {
       "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. JSONデータをさまざまなモードでインポートします。

   • シンプルモードでJSONデータをインポートします。サンプルコード:

     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ブローカーリスト
         "kafka_topic" = "my_topic", // Kafkaトピック
         "kafka_partitions" = "0,1,2", // Kafkaパーティション
         "kafka_offsets" = "0,0,0" // Kafkaオフセット
      );

   • JSON データを正確にインポートします。サンプルコード:

     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\"]", // JSONパス
         "strip_outer_array" = "true" // 外側の配列を削除
     )
     FROM KAFKA
     (
         "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092", // Kafkaブローカーリスト
         "kafka_topic" = "my_topic", // Kafkaトピック
         "kafka_partitions" = "0,1,2", // Kafkaパーティション
         "kafka_offsets" = "0,0,0" // Kafkaオフセット
     );

     説明

     サンプルデータでは、テーブルのパーティションフィールド dt は使用できません。 dt フィールドの値は、CREATE ROUTINE LOAD ステートメントの dt=from_unixtime(timestamp,'%Y%m%d') 構成を使用して生成されます。

さまざまな認証方式を使用する Kafka クラスターへのアクセス

次の例は、Kafkaクラスターの認証方式に基づいてKafkaクラスターにアクセスする方法を示しています。

1. SSL認証方式が有効になっているKafkaクラスタにアクセスします。

   SSL認証方式が有効になっているKafkaクラスタにアクセスするには、Kafkaブローカーの公開鍵を認証するために使用される証明書ファイル（ca.pem）を提供する必要があります。 Kafkaクラスタでクライアント認証モードが有効になっている場合は、クライアントの公開鍵証明書（client.pem）、秘密鍵ファイル（client.key）、および秘密鍵のパスワードも必要です。必要なファイルは、CREATE FILE ステートメントを実行して、事前にApsaraDB for SelectDBにアップロードする必要があります。カタログ名は kafka です。

   1. ファイルをアップロードします。サンプルコード：

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

   2. ルーチンロードジョブを作成します。サンプルコード：

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

      説明

      ApsaraDB for SelectDBは、Kafka C++クライアントライブラリ librdkafka を使用してKafkaクラスタにアクセスします。 librdkafkaでサポートされているパラメータの詳細については、librdkafka のConfiguration properties ドキュメントをご参照ください。

2. PLAIN認証方式が有効になっているKafkaクラスタにアクセスします。

   PLAIN認証方式が有効になっているKafkaクラスタにアクセスするには、次の設定を追加する必要があります。

   1. property.security.protocol=SASL_PLAINTEXT: Simple Authentication and Security Layer（SASL）プレーンテキスト認証方式を使用します。

   2. property.sasl.mechanism=PLAIN: SASL認証方式をPLAINに設定します。

   3. property.sasl.username=admin: SASLのユーザー名を指定します。

   4. property.sasl.password=admin: SASLのパスワードを指定します。

   ルーチンロードジョブを作成します。サンプルコード：

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

3. Kerberos認証方式が有効になっているKafkaクラスタにアクセスします。

   Kerberos認証方式が有効になっているKafkaクラスタにアクセスするには、次の設定を追加する必要があります。

   1. security.protocol=SASL_PLAINTEXT: SASLプレーンテキスト認証方式を使用します。

   2. sasl.kerberos.service.name=$SERVICENAME: ブローカーサービス名を指定します。

   3. sasl.kerberos.keytab=/etc/security/keytabs/${CLIENT_NAME}.keytab: ローカルの.keytabファイルのパスを指定します。

   4. sasl.kerberos.principal=${CLIENT_NAME}/${CLIENT_HOST}: ApsaraDB for SelectDBがKafkaクラスタへの接続に使用するKerberosプリンシパルを指定します。

   ルーチンロードジョブを作成します。サンプルコード：

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

   説明

   • ApsaraDB for SelectDBがKerberos認証方式が有効になっているKafkaクラスタにアクセスできるようにするには、ApsaraDB for SelectDBクラスタのすべての実行ノードにKerberosクライアント kinit をデプロイし、krb5.confファイルを構成し、Key Distribution Center（KDC）サービス情報を指定する必要があります。

   • property.sasl.kerberos.keytab パラメータをローカルの.keytabファイルの絶対パスに設定し、ApsaraDB for SelectDBプロセスがローカルの.keytabファイルにアクセスできるようにします。

ALTER ROUTINE LOAD FOR <job_name>
[job_properties]
FROM <data_source>
[data_source_properties]

PAUSE [ALL] ROUTINE LOAD FOR <job_name>;

RESUME [ALL] ROUTINE LOAD FOR <job_name>

STOP ROUTINE LOAD FOR <job_name>;

STOP ROUTINE LOAD FOR test1;

SHOW [ALL] ROUTINE LOAD [FOR job_name];