CREATE ROUTINE LOAD
功能
Routine Load 支持持续消费 Apache Kafka® 的消息并导入至 StarRocks 中。Routine Load 支持 Kafka 中消息的格式为 CSV、JSON、Avro (自 v3.0.1),并且访问 Kafka 时,支持多种安全协议,包括 plaintext
、ssl
、sasl_plaintext
和 sasl_ssl
。
本文介绍 CREATE ROUTINE LOAD 的语法、参数说明和示例。
说明
- Routine Load 的应用场景、基本原理和基本操作,请参见 使用 Routine Load 导入数据。
- Routine Load 操作需要目标表的 INSERT 权限。如果您的用户账号没有 INSERT 权限,请参考 GRANT 给用户赋权。
语法
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 。默认值:1 (表示实际不会生效)。建议您将该值设置为 0 。这样的话,当导入的数据行中有错误时,导入作业会暂停,从而保证数据的正确性。如果希望忽略错误的数据行,可以设置该参数的取值大于 0 。这样的话,即使导入的数据行中有错误,导入作业也能成功。注意
|
CREATE_ROUTINE_LOAD.md | ||
strict_mode | 否 | 是否开启严格模式。取值范围:TRUE 或者 FALSE 。默认值:FALSE 。开启后,如果源数据某列的值为 NULL ,但是目标表中该列不允许为 NULL ,则该行数据会被过滤掉。关于该模式的介绍,参见严格模式。 |
log_rejected_record_num | 否 | 指定最多允许记录多少条因数据质量不合格而过滤掉的数据行数。该参数自 3.1 版本起支持。取值范围:0 、-1 、大于 0 的正整数。默认值:0 。
|
timezone | 否 | 该参数的取值会影响所有导入涉及的、跟时区设置有关的函数所返回的结果。受时区影响的函数有 strftime、alignment_timestamp 和 from_unixtime 等,具体请参见设置时区。导入参数 timezone 设置的时区对应设置时区中所述的会话级时区。 |
merge_condition | 否 | 用于指定作为更新生效条件的列名。这样只有当导入的数据中该列的值大于等于当前值的时候,更新才会生效。参见通过导入实现数据变更。指定的列必须为非主键列,且仅主键模型表支持条件更新。 |
strict_mode | 否 | 是否开启严格模式。取值范围:TRUE 或者 FALSE 。默认值:FALSE 。开启后,如果源数据某列的值为 NULL ,但是目标表中该列不允许为 NULL ,则该行数据会被过滤掉。关于该模式的介绍,参见严格模式。 |
log_rejected_record_num | 否 | 指定最多允许记录多少条因数据质量不合格而过滤掉的数据行数。该参数自 3.1 版本起支持。取值范围:0 、-1 、大于 0 的正整数。默认值:0 。
|
timezone | 否 | 该参数的取值会影响所有导入涉及的、跟时区设置有关的函数所返回的结果。受时区影响的函数有 strftime、alignment_timestamp 和 from_unixtime 等,具体请参见设置时区。导入参数 timezone 设置的时区对应设置时区中所述的会话级时区。 |
merge_condition | 否 | 用于指定作为更新生效条件的列名。这样只有当导入的数据中该列的值大于等于当前值的时候,更新才会生效。参见通过导入实现数据变更。指定的列必须为非主键列,且仅主键模型表支持条件更新。 |
format | 否 | 源数据的格式,取值范围:CSV 、JSON 或者 Avro (自 v3.0.1)。默认值:CSV 。 |
trim_space | 否 | 用于指定是否去除 CSV 文件中列分隔符前后的空格。取值类型:BOOLEAN。默认值:false 。有些数据库在导出数据为 CSV 文件时,会在列分隔符的前后添加一些空格。根据位置的不同,这些空格可以称为“前导空格”或者“尾随空格”。通过设置该参数,可以使 StarRocks 在导入数据时删除这些不必要的空格。 需要注意的是,StarRocks 不会去除被 enclose 指定字符 括起来的字段内的空格(包括字段的前导空格和尾随空格)。例如,列分隔符是竖线 (| ),enclose 指定的字符是双引号 (" ):| "Love StarRocks" | 。如果设置 trim_space 为 true,则 StarRocks 处理后的结果数据为 |"Love StarRocks"| 。 |
enclose | 否 | 根据 RFC4180,用于指定把 CSV 文件中的字段括起来的字符。取值类型:单字节字符。默认值:NONE 。最常用 enclose 字符为单引号 (' ) 或双引号 (" )。被 enclose 指定字符括起来的字段内的所有特殊字符(包括行分隔符、列分隔符等)均看做是普通符号。比 RFC4180 标准更进一步的是,StarRocks 提供的 enclose 属性支持设置任意单个字节的字符。如果一个字段内包含了 enclose 指定字符,则可以使用同样的字符对 enclose 指定字符进行转义。例如,在设置了enclose 为双引号 (" ) 时,字段值 a "quoted" c 在 CSV 文件中应该写作 "a ""quoted"" c" 。 |
escape | 否 | 指定用于转义的字符。用来转义各种特殊字符,比如行分隔符、列分隔符、转义符、enclose 指定字符等,使 StarRocks 把这些特殊字符当做普通字符而解析成字段值的一部分。取值类型:单字节字符。默认值:NONE 。最常用的 escape 字符为斜杠 (\ ),在 SQL 语句中应该写作双斜杠 (\\ )。escape 指定字符同时作用于 enclose 指定字符 的内部和外部。以下为两个示例:
|
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 数据的根节点。 |
task_consume_second | 否 | 单个 Routine Load 导入作业中每个 Routine Load 导入任务消费数据的最大时长,单位为秒。相较于FE 动态参数 routine_load_task_consume_second (作用于集群内部所有 Routine Load 导入作业),该参数仅针对单个 Routine Load 导入作业,更加灵活。该参数自 v3.1.0 起新增。
|
task_timeout_second | 否 | Routine Load 导入作业中每个 Routine Load 导入任务超时时间,单位为秒。相较于FE 动态参数 routine_load_task_timeout_second (作用于集群内部所有 Routine Load 导入作业),该参数仅针对单个 Routine Load 导入作业,更加灵活。该参数自 v3.1.0 起新增。
|
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 |