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任务。导入任务为严格模式，示例如下。

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