CREATE ROUTINE LOAD
功能
Routine Load 支持持续消费 Apache Kafka® 的消息并导入至 StarRocks 中。Routine Load 支持 Kafka 中消息的格式为 CSV 和 JSON,并且并且访问 Kafka 时,支持多种安全协议,包括 plaintext
、ssl
、sasl_plaintext
和 sasl_ssl
。
本文介绍 CREATE ROUTINE LOAD 的语法、参数说明和示例。
说明
Routine Load 的应用场景、基本原理和基本操作,请参见 从 Apache Kafka® 持续导入。
语法
CREATE ROUTINE LOAD <database_name>.<job_name> ON <table_name>
[load_properties]
[job_properties]
FROM data_source
[data_source_properties]
参数说明
database_name
、job_name
、table_name
database_name
选填,目标数据库的名称。
job_name
必填,导入作业的名称。一张表可能有多个导入作业,建议您利用具有辨识度的信息(例如 Kafka Topic 名称、创建导入作业的大致时间等)来设置具有意义的导入作业名称,用于区分多个导入作业。同一数据库内,导入作业的名称必须唯一。
table_name
必填,目标表的名称。
load_properties
选填。源数据的属性。语法:
[COLUMNS TERMINATED BY '<column_separator>'],
[ROWS TERMINATED BY '<row_separator>'],
[COLUMNS (<column1_name>[,<column2_name>,<column_assignment>,... ])],
[WHERE <expr>],
[PARTITION (<partition1_name>[,<partition2_name>,...])]
[TEMPORARY PARTITION (<temporary_partition1_name>[,<temporary_partition2_name>,...])]
如果导入 CSV 格式的数据,则可以指定列分隔符,默认为\t
,即 Tab。例如可以输入 COLUMNS TERMINATED BY ","
。指定列分隔符为逗号(,)。
说明
- 必须确保这里指定的列分隔符与源数据中的列分隔符一致。
- StarRocks 支持设置长度最大不超过 50 个字节的 UTF-8 编码字符串作为列分隔符,包括常见的 逗号 (,)、Tab 和 Pipe (|)。
- 空值 (null) 用
\N
表示。比如,源数据一共有三列,其中某行数据的第一列、第三列数据分别为a
和b
,第二列没有数据,则第二列需要用\N
来表示空值,写作a,\N,b
,而不是a,,b
。a,,b
表示第二列是一个空字符串。
ROWS TERMINATED BY
用于指定源数据中的行分隔符。如果不指定该参数,则默认为 \n
。
COLUMNS
源数据和目标表之间的列映射和转换关系。详细说明,请参见列映射和转换关系。
column_name
:映射列,源数据中这类列的值可以直接落入目标表的列中,不需要进行计算。column_assignment
:衍生列,格式为column_name = expr
,源数据中这类列的值需要基于表达式expr
进行计算后,才能落入目标表的列中。 建议将衍生列排在映射列之后,因为 StarRocks 先解析映射列,再解析衍生列。
说明
以下情况不需要设置
COLUMNS
参数:
- 待导入 CSV 数据中的列与目标表中列的数量和顺序一致。
- 待导入 JSON 数据中的 Key 名与目标表中的列名一致。
WHERE
设置过滤条件,只有满足过滤条件的数据才会导入到 StarRocks 中。例如只希望导入 col1
大于 100
并且 col2
等于 1000
的数据行,则可以输入 WHERE col1 > 100 and col2 = 1000
。
说明
过滤条件中指定的列可以是源数据中本来就存在的列,也可以是基于源数据的 列生成的衍生列。
PARTITION
将数据导入至目标表的指定分区中。如果不指定分区,则会将数据自动导入至其对应的分区中。 示例:
PARTITION(p1, p2, p3)
job_properties
必填。导入作业的属性。语法:
PROPERTIES ("<key1>" = "<value1>"[, "<key2>" = "<value2>" ...])
参数说明如下:
参数 | 是否必选 | 说明 |
---|---|---|
desired_concurrent_number | 否 | 单个 Routine Load 导入作业的期望任务并发度,表示期望一个导入作业最多被分成多少个任务并行执行。默认值为 3 。 但是实际任务并行度由如下多个参数组成的公式决定,并且实际任务并行度的上限为 BE 节点的数量或者消费分区的数量。min(alive_be_number, partition_number, desired_concurrent_number, max_routine_load_task_concurrent_num) 。
|
max_batch_interval | 否 | 任务的调度间隔,即任务多久执行一次。单位:秒。取值范围:5 ~60 。默认值:10 。建议取值为导入间隔 10s 以上,否则会因为导入频率过高可能会报错版本数过多。 |
max_batch_rows | 否 | 该参数只用于定义错误检测窗口范围,错误检测窗口范围为单个 Routine Load 导入任务所消费的 10 * max-batch-rows 行数据,默认为 10 * 200000 = 2000000 。导入任务时会检测窗口中数据是否存在错误。错误数据是指 StarRocks 无法解析的数据,比如非法的 JSON。 |
max_error_number | 否 | 错误检测窗口范围内允许的数据行数的上限。当错误数据行数超过该值时,导入作业会暂停,此时您需要执行 SHOW ROUTINE LOAD,根据 ErrorLogUrls ,检查 Kafka 中的消息并且更正错误。默认为 0 ,表示不允许有错误行。注意
|
max_filter_ratio | 否 | 用于指定导入作业的最大容错率,即导入作业能够容忍的因数据质量不合格而过滤掉的数据行所占的最大比例。取值范围:0 ~1 。默认值:0 。建议您保留默认值 0 。这样的话,当导入的数据行中有错误时,导入作业会暂停,从而保证数据的正确性。如果希望忽略错误的数据行,可以设置该参数的取值大于 0 。这样的话,即使导入的数据行中有错误,导入作业也能成功。注意
|
strict_mode | 否 | 是否开启严格模式。取值范围:TRUE 或者 FALSE 。默认值:FALSE 。开启后,如果源数据某列的值为 NULL ,但是目标表中该列不允许为 NULL ,则该行数据会被过滤掉。关于该模式的介绍,参见严格模式。 |
timezone | 否 | 该参数的取值会影响所有导入涉及的、跟时区设置有关的函数所返回的结果。受时区影响的函数有 strftime、alignment_timestamp 和 from_unixtime 等,具体请参见设置时区。导入参数 timezone 设置的时区对应设置时区中所述的会话级时区。 |
merge_condition | 否 | 用于指定作为更新生效条件的列名。这样只有当导入的数据中该列的值大于等于当前值的时候,更新才会生效。参见通过导入实现数据变更。指定的列必须为非主键列,且仅主键模型表支持条件更新。 |
format | 否 | 源数据的格式,取值范围:CSV 或者 JSON 。默认值:CSV 。 |
strip_outer_array | 否 | 是否裁剪 JSON 数据最外层的数组结构。取值范围:TRUE 或者 FALSE 。默认值:FALSE 。真实业务场景中,待导入的 JSON 数据可能在最外层有一对表示数组结构的中括号 [] 。这种情况下,一般建议您指定该参数取值为 true ,这样 StarRocks 会剪裁掉外层的中括号 [] ,并把中括号 [] 里的每个内层数组都作为一行单独的数据导入。如果您指定该参数取值为 false ,则 StarRocks 会把整个 JSON 数据解析成一个数组,并作为一行数据导入。例如,待导入 的 JSON 数据为 [ {"category" : 1, "author" : 2}, {"category" : 3, "author" : 4} ] ,如果指定该参数取值为 true ,则 StarRocks 会把 {"category" : 1, "author" : 2} 和 {"category" : 3, "author" : 4} 解析成两行数据,并导入到目标表中对应的数据行。 |
jsonpaths | 否 | 用于指定待导入的字段的名称。仅在使用匹配模式导入 JSON 数据时需要指定该参数。参数取值为 JSON 格式。参见目标表存在衍生列,其列值通过表达式计算生成。 |
json_root | 否 | 如果不需要导入整个 JSON 数据,则指定实际待导入 JSON 数据的根节点。参数取值为合法的 JsonPath。默认值为空,表示会导入整个 JSON 数据。具体请参见本文提供的示例指定实际待导入 JSON 数据的根节点。 |
data_source
、data_source_properties
数据源和数据源属性。语法:
FROM <data_source>
("<key1>" = "<value1>"[, "<key2>" = "<value2>" ...])
data_source
必填。指定数据源,目前仅支持取值为 KAFKA
。
data_source_properties
必填。数据源属性,参数以及说明如下:
参数 | 说明 |
---|---|
kafka_broker_list | Kafka 的 Broker 连接信息。格式为 <kafka_broker_ip>:<kafka port> ,多个 Broker 之间以英文逗号 (,) 分隔。 Kafka Broker 默认端口号为 9092 。示例:"kafka_broker_list" = "xxx.xx.xxx.xx:9092,xxx.xx.xxx.xx:9092" |
kafka_topic | Kafka Topic 名称。一个导入作业仅支持消费一个 Topic 的消息。 |
kafka_partitions | 待消费的分区。示例:"kafka_partitions" = "0, 1, 2, 3" 。如果不配置该参数,则默认消费所有分区。 |
kafka_offsets | 待消费分区的起始消费位点,必须一一对应 kafka_partitions 中指定的每个分区。如果不配置该参数,则默认为从分区的末尾开始消费。支持取值为
示例: "kafka_offsets" = "1000, OFFSET_BEGINNING, OFFSET_END, 2000" 。 |
property.kafka_default_offsets | 所有待消费分区的默认起始消费位点。支持的取值与 kafka_offsets 一致。 |
更多数据源相关参数
支持设置更多数据源 Kafka 相关参数,功能等同于 Kafka 命令行 --property
, 支持参数,请参见 librdkafka 配置项文档中适用于客户端的配置项。
说明
当参数的取值是文件时,则值前加上关键词
FILE:
。关于如何创建文件,请参见 CREATE FILE 命令文档。
指定所有待消费分区的默认起始消费位点。
"property.kafka_default_offsets" = "OFFSET_BEGINNING"
property.kafka_default_offsets
的取值为具体的消费位点,或者:
OFFSET_BEGINNING
:从分区中有数据的位置开始消费。OFFSET_END
:从分区的末尾开始消费。
指定导入任务消费 Kafka 时所基于 Consumer Group 的 group.id
"property.group.id" = "group_id_0"
如果没有指定 group.id
,StarRocks 会根据 Routine Load 的导入作业名称生成一个随机值,具体格式为{job_name}_{random uuid}
,如 simple_job_0a64fe25-3983-44b2-a4d8-f52d3af4c3e8
。
指定 BE 访问 Kafka 时的安全协议并配置相关参数
支持安全协议为 plaintext
(默认)、ssl
、sasl_plaintext
和 sasl_ssl
,并且需要根据安全协议配置相关参数。
当安全协议为 sasl_plaintext
或 sasl_ssl
时,支持如下 SASL 认证机制:
- PLAIN
- SCRAM-SHA-256 和 SCRAM-SHA-512
- OAUTHBEARER
示例:
-
访问 Kafka 时,使用安全协议 SSL
"property.security.protocol" = "ssl", -- 指定安全协议为 SSL
"property.ssl.ca.location" = "FILE:ca-cert", -- CA 证书的位置
--如果 Kafka server 端开启了 client 认证,则还需设置如下三个参数:
"property.ssl.certificate.location" = "FILE:client.pem", -- Client 的 public key 的位置
"property.ssl.key.location" = "FILE:client.key", -- Client 的 private key 的位置
"property.ssl.key.password" = "abcdefg" -- Client 的 private key 的密码 -
访问 Kafka 时,使用 SASL_PLAINTEXT 安全协议和 SASL/PLAIN 认证机制
"property.security.protocol" = "SASL_PLAINTEXT", -- 指定安全协议为 SASL_PLAINTEXT
"property.sasl.mechanism" = "PLAIN", -- 指定 SASL 认证机制为 PLAIN
"property.sasl.username" = "admin", -- SASL 的用户名
"property.sasl.password" = "admin" -- SASL 的密码
FE 和 BE 配置项
Routine Load 相关配置项,请参见配置参数。
列映射和转换关系
导入 CSV 数据
如果 CSV 格式的数据中的列与目标表中的列的数量或顺序不一致,则需要通过 COLUMNS
参数来指定源数据和目标表之间的列映射和转换关系。一般包括如下两种场景:
-
源数据中的列与目标表中的列数量一致,但是顺序不一致。并且数据不需要通过函数计算、可以直接落入目标表中对应的列。 您需要在
COLUMNS
参数中按照源数据中的列顺序、使用目标表中对应的列名来配置列映射和转换关系。例如,目标表中有三列,按顺序依次为
col1
、col2
和col3
;源数据中也有三列,按顺序依次对应目标表中的col3
、col2
和col1
。这种情况下,需要指定COLUMNS(col3, col2, col1)
。 -
源数据中的列与目标表中的列数量不一致,甚至某些列的数据需要通过转换(函数计算以后)才能落入目标表中对应的列。 您不仅需要在
COLUMNS
参数中按照源数据中的列顺序、使用目标表中对应的列名来配置列映射关系,还需要指定参与数据计算的函数。以下为两个示例:- 源数据比目标表多列。
比如目标表中有三列,按顺序依次为
col1
、col2
和col3
;源数据中有四列,前三列按顺序依次对应目标表中的col1
、col2
和col3
,第四列在目标表中无对应的列。这种情况下,需要指定COLUMNS(col1, col2, col3, temp)
,其中,最后一列可随意指定一个名称(如temp
)用于占位即可。 - 目标表存在基于源数据的列进行计算后生成的衍生列
- 源数据比目标表多列。
比如目标表中有三列,按顺序依次为