創(chuàng)建導(dǎo)入作業(yè)

使用Routine Load功能時(shí)，首先需創(chuàng)建一個(gè)Routine Load作業(yè)。該作業(yè)將通過(guò)例行調(diào)度持續(xù)發(fā)送任務(wù)，每個(gè)任務(wù)會(huì)消耗一定數(shù)量的Kafka消息。

語(yǔ)法

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

參數(shù)說(shuō)明

[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"

使用示例

創(chuàng)建Routine Load簡(jiǎn)單作業(yè)

1. 創(chuàng)建待導(dǎo)入的SelectDB數(shù)據(jù)表，示例如下。

   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. 分別設(shè)置不同的參數(shù)導(dǎo)入數(shù)據(jù)，示例如下。

   • 為example_db的test_table創(chuàng)建一個(gè)名為test1的Kafka Routine Load任務(wù)。指定列分隔符的group.id和client.id，設(shè)置自動(dòng)默認(rèn)消費(fèi)所有分區(qū)，且從有數(shù)據(jù)的位置（OFFSET_BEGINNING）開(kāi)始訂閱，示例如下。

     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創(chuàng)建一個(gè)名為test2的Kafka Routine Load任務(wù)。導(dǎo)入任務(wù)為嚴(yán)格模式，示例如下。

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

   • 從指定的時(shí)間點(diǎn)開(kāi)始消費(fèi)，示例如下。

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

導(dǎo)入JSON格式數(shù)據(jù)

Routine Load導(dǎo)入的JSON格式僅支持以下兩種。

• 只有一條記錄，且為JSON對(duì)象。

  當(dāng)使用單表導(dǎo)入（即通過(guò)ON TABLE_NAME指定表名）時(shí)，JSON數(shù)據(jù)格式如下。

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

  當(dāng)使用動(dòng)態(tài)或多表導(dǎo)入Routine Load（即不指定具體的表名）時(shí)，JSON數(shù)據(jù)格式如下。

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

• JSON數(shù)組，數(shù)組中可含多條記錄。

  當(dāng)使用單表導(dǎo)入（即通過(guò)ON TABLE_NAME指定表名）時(shí)，JSON數(shù)據(jù)格式如下。

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

  當(dāng)使用動(dòng)態(tài)或多表導(dǎo)入（即不指定具體的表名）時(shí)，JSON數(shù)據(jù)格式如下。

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

導(dǎo)入JSON格式數(shù)據(jù)，示例如下。

1. 創(chuàng)建待導(dǎo)入的SelectDB數(shù)據(jù)表，示例如下。

   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. 分別使用兩個(gè)類(lèi)型的JSON數(shù)據(jù)記錄到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. 以不同模式導(dǎo)入JSON數(shù)據(jù)，示例如下。

   • 以簡(jiǎn)單模式導(dǎo)入JSON數(shù)據(jù)。

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

   • 精準(zhǔn)導(dǎo)入JSON格式數(shù)據(jù)。

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

     說(shuō)明

     表里的分區(qū)字段dt在示例數(shù)據(jù)里并沒(méi)有，而是在Routine Load語(yǔ)句里通過(guò)dt=from_unixtime(timestamp,'%Y%m%d')轉(zhuǎn)換出來(lái)的。

訪問(wèn)不同驗(yàn)證方式的Kafka集群

根據(jù)Kafka集群驗(yàn)證方式的不同，訪問(wèn)的方式示例如下。

1. 訪問(wèn)SSL認(rèn)證的Kafka集群。

   訪問(wèn)SSL認(rèn)證的Kafka集群需要您提供用于認(rèn)證Kafka Broker公鑰的證書(shū)文件（ca.pem）。如果Kafka集群同時(shí)開(kāi)啟了客戶端認(rèn)證，則還需提供客戶端的公鑰（client.pem）、密鑰文件（client.key），以及密鑰密碼。這里所需的文件需要先通過(guò)CREATE FILE命令上傳到SelectDB中，并且Catalog名稱(chēng)為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. 創(chuàng)建Routine Load作業(yè)，示例如下。

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

      說(shuō)明

      SelectDB通過(guò)Kafka的C++ APIlibrdkafka來(lái)訪問(wèn)Kafka集群。librdkafka所支持的參數(shù)可以參閱Configuration properties。

2. 訪問(wèn)PLAIN認(rèn)證的Kafka集群。

   訪問(wèn)開(kāi)啟PLAIN認(rèn)證的Kafka集群，需要增加配置如下。

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

   2. property.sasl.mechanism=PLAIN：設(shè)置SASL的認(rèn)證方式為PLAIN。

   3. property.sasl.username=admin：設(shè)置SASL的用戶名。

   4. property.sasl.password=admin：設(shè)置SASL的密碼。

   創(chuàng)建Routine Load作業(yè)，示例如下。

   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. 訪問(wèn)Kerberos認(rèn)證的Kafka集群。

   訪問(wèn)開(kāi)啟kerberos認(rèn)證的Kafka集群，需要增加配置如下。

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

   2. sasl.kerberos.service.name=$SERVICENAME：設(shè)置broker servicename。

   3. sasl.kerberos.keytab=/etc/security/keytabs/${CLIENT_NAME}.keytab：設(shè)置Keytab本地文件路徑。

   4. sasl.kerberos.principal=${CLIENT_NAME}/${CLIENT_HOST}：設(shè)置SelectDB連接Kafka時(shí)使用的Kerberos主體。

   創(chuàng)建Routine Load作業(yè)，示例如下。

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

   說(shuō)明

   • 若要使SelectDB訪問(wèn)開(kāi)啟kerberos認(rèn)證方式的Kafka集群，需要在SelectDB集群所有運(yùn)行節(jié)點(diǎn)上部署Kerberos客戶端kinit，并配置krb5.conf，填寫(xiě)KDC服務(wù)信息等。

   • 配置property.sasl.kerberos.keytab的值需要指定Keytab本地文件的絕對(duì)路徑，并允許SelectDB進(jìn)程訪問(wèn)該本地文件。

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