ApsaraDB for SelectDB：Routine Load

建立匯入作業

使用Routine Load功能時，首先需建立一個Routine Load作業。該作業將通過例行調度持續發送任務，每個任務會消耗一定數量的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"

使用樣本

建立Routine Load簡單作業

1. 建立待匯入的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的Kafka Routine Load任務。指定資料行分隔符號的group.id和client.id，設定自動預設消費所有分區，且從有資料的位置（OFFSET_BEGINNING）開始訂閱，樣本如下。

     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的Kafka Routine Load任務。匯入任務為strict 模式，樣本如下。

     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匯入的JSON格式僅支援以下兩種。

• 只有一條記錄，且為JSON對象。

  當使用單表匯入（即通過ON TABLE_NAME指定表名）時，JSON資料格式如下。

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

  當使用動態或多表匯入Routine Load（即不指定具體的表名）時，JSON資料格式如下。

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

• JSON數組，數組中可含多條記錄。

  當使用單表匯入（即通過ON TABLE_NAME指定表名）時，JSON資料格式如下。

  [
      {   
          "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. 建立待匯入的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"
   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. 分別使用兩個類型的JSON資料記錄到Topic裡：

   {
       "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_topic" = "my_topic",
         "kafka_partitions" = "0,1,2",
         "kafka_offsets" = "0,0,0"
      );

   • 精準匯入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\"]",
         "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"
     );

     說明

     表裡的分區欄位dt在樣本資料裡並沒有，而是在Routine Load語句裡通過dt=from_unixtime(timestamp,'%Y%m%d')轉換出來的。

訪問不同驗證方式的Kafka叢集

根據Kafka叢集驗證方式的不同，訪問的方式樣本如下。

1. 訪問SSL認證的Kafka叢集。

   訪問SSL認證的Kafka叢集需要您提供用於認證Kafka Broker公開金鑰的認證檔案（ca.pem）。如果Kafka叢集同時開啟了用戶端認證，則還需提供用戶端的公開金鑰（client.pem）、密鑰檔案（client.key），以及密鑰密碼。這裡所需的檔案需要先通過CREATE FILE命令上傳到SelectDB中，並且Catalog名稱為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. 建立Routine Load作業，樣本如下。

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

      說明

      SelectDB通過Kafka的C++ APIlibrdkafka來訪問Kafka叢集。librdkafka所支援的參數可以參閱Configuration properties。

2. 訪問PLAIN認證的Kafka叢集。

   訪問開啟PLAIN認證的Kafka叢集，需要增加配置如下。

   1. property.security.protocol=SASL_PLAINTEXT：使用SASL plaintext。

   2. property.sasl.mechanism=PLAIN：設定SASL的認證方式為PLAIN。

   3. property.sasl.username=admin：設定SASL的使用者名稱。

   4. property.sasl.password=admin：設定SASL的密碼。

   建立Routine Load作業，樣本如下。

   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 plaintext。

   2. sasl.kerberos.service.name=$SERVICENAME：設定broker servicename。

   3. sasl.kerberos.keytab=/etc/security/keytabs/${CLIENT_NAME}.keytab：設定Keytab本地檔案路徑。

   4. sasl.kerberos.principal=${CLIENT_NAME}/${CLIENT_HOST}：設定SelectDB串連Kafka時使用的Kerberos主體。

   建立Routine Load作業，樣本如下。

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

   說明

   • 若要使SelectDB訪問開啟kerberos認證方式的Kafka叢集，需要在SelectDB叢集所有運行節點上部署Kerberos用戶端kinit，並配置krb5.conf，填寫KDC服務資訊等。

   • 配置property.sasl.kerberos.keytab的值需要指定Keytab本地檔案的絕對路徑，並允許SelectDB進程訪問該本地檔案。

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];