Doris-Routine Load（二十七）

例行导入（Routine Load）功能为用户提供了一种自动从指定数据源进行数据导入的功能。

适用场景

当前仅支持从 Kafka 系统进行例行导入，使用限制：

（1）支持无认证的 Kafka 访问，以及通过 SSL 方式认证的 Kafka 集群。

（2）支持的消息格式为 csv, json 文本格式。csv 每一个 message 为一行，且行尾不包含换行符。

（3）默认支持 Kafka 0.10.0.0（含）以上版本。如果要使用 Kafka 0.10.0.0 以下版本(0.9.0, 0.8.2, 0.8.1, 0.8.0)，需要修改 be 的配置，将 kafka_broker_version_fallback 的值设置为要兼容的旧版本，或者在创建 routine load 的时候直接设置property.broker.version.fallback的值为要兼容的旧版本，使用旧版本的代价是 routine load 的部分新特性可能无法使用，如根据时间设置 kafka 分区的 offset。

基本原理

如上图，Client 向 FE 提交一个例行导入作业。

（1）FE 通过 JobScheduler 将一个导入作业拆分成若干个 Task。每个 Task 负责导入指定的一部分数据。Task 被 TaskScheduler 分配到指定的 BE 上执行。

（2）在 BE 上，一个 Task 被视为一个普通的导入任务，通过 Stream Load 的导入机制进行导入。导入完成后，向 FE 汇报。

（3）FE 中的 JobScheduler 根据汇报结果，继续生成后续新的 Task，或者对失败的Task 进行重试。

（4）整个例行导入作业通过不断的产生新的 Task，来完成数据不间断的导入。

基本语法

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

执行 HELP ROUTINE LOAD 可以查看语法帮助，下面是参数说明

1）[db.]job_name

导入作业的名称，在同一个 database 内，相同名称只能有一个 job 在运行。

2）tbl_name

指定需要导入的表的名称。

3）merge_type

数据的合并类型，一共支持三种类型 APPEND、DELETE、MERGE 其中，APPEND 是默认值，表示这批数据全部需要追加到现有数据中，DELETE 表示删除与这批数据 key 相同的所有行，MERGE 语义 需要与 delete on 条件联合使用，表示满足 delete 条件的数据按照 DELETE 语义处理其余的按照 APPEND 语义处理 , 语法为

[WITHMERGE|APPEND|DELETE]

4）load_properties

用于描述导入数据。语法：

[column_separator], [columns_mapping], [where_predicates], [delete_on_predicates], [source_sequence], [partitions], [preceding_predicates]

（1）column_separator:

指定列分隔符，如： COLUMNS TERMINATED BY "," 这个只在文本数据导入的时候需要指定，JSON 格式的数据导入不需要指定这个参数。

默认为：\t

（2）columns_mapping:

指定源数据中列的映射关系，以及定义衍生列的生成方式。

映射列：

按顺序指定，源数据中各个列，对应目的表中的哪些列。对于希望跳过的列，可以指定一个不存在的列名。假设目的表有三列 k1, k2, v1。源数据有 4 列，其中第 1、2、4 列分别对应 k2, k1, v1。则书写如下：

COLUMNS (k2, k1, xxx, v1)

其中 xxx 为不存在的一列，用于跳过源数据中的第三列。

衍生列：

以 col_name = expr 的形式表示的列，我们称为衍生列。即支持通过 expr 计算得出目的表中对应列的值。 衍生列通常排列在映射列之后，虽然这不是强制的规定，但是 Doris 总是先解析映射列，再解析衍生列。 接上一个示例，假设目的表还有第 4 列 v2，v2 由 k1 和 k2 的和产生。则可以书写如下：

COLUMNS (k2, k1, xxx, v1, v2 = k1 + k2);

再举例，假设用户需要导入只包含 k1 一列的表，列类型为 int。并且需要将源文件中的对应列进行处理：将负数转换为正数，而将正数乘以 100。这个功能可以通过 case when 函数实现，正确写法应如下：

COLUMNS (xx, k1 = case when xx < 0 then cast(-xx as varchar) else cast((xx + '100') as varchar) end)

（3）where_predicates

用于指定过滤条件，以过滤掉不需要的列。过滤列可以是映射列或衍生列。 例如我们只希望导入 k1 大于 100 并且 k2 等于 1000 的列，则书写如下：

WHERE k1 > 100 and k2 = 1000

（4）partitions

指定导入目的表的哪些 partition 中。如果不指定，则会自动导入到对应的 partition 中。

示例：

PARTITION(p1, p2, p3)

（5）delete_on_predicates

表示删除条件，仅在 merge type 为 MERGE 时有意义，语法与 where 相同

（6）source_sequence:

只适用于 UNIQUE_KEYS,相同 key 列下，保证 value 列按照 source_sequence 列进行REPLACE, source_sequence 可以是数据源中的列，也可以是表结构中的一列。

（7）preceding_predicates

PRECEDING FILTER predicate用于过滤原始数据。原始数据是未经列映射、转换的数据。用户可以在对转换前的数据前进行一次过滤，选取期望的数据，再进行转换。

5）job_properties

用于指定例行导入作业的通用参数。 语法：

PROPERTIES ("key1" = "val1","key2" = "val2"
)

目前支持以下参数：

（1）desired_concurrent_number

期望的并发度。一个例行导入作业会被分成多个子任务执行。这个参数指定一个作业最多有多少任务可以同时执行。必须大于 0。默认为 3。 这个并发度并不是实际的并发度，实际的并发度，会通过集群的节点数、负载情况，以及数据源的情况综合考虑。

一个作业，最多有多少 task 同时在执行。对于 Kafka 导入而言，当前的实际并发度计算如下：

Min(partition num, desired_concurrent_number, alive_backend_num, 
Config.max_routine_load_task_concurrrent_num)

其中 Config.max_routine_load_task_concurrrent_num 是系统的一个默认的最大并发数限制。这是一个 FE 配置，可以通过改配置调整。默认为 5。

其中 partition num 指订阅的 Kafka topic 的 partition 数量。alive_backend_num 是当前正常的 BE 节点数。

（2）max_batch_interval/max_batch_rows/max_batch_size这三个参数分别表示：

• ① 每个子任务最大执行时间，单位是秒。范围为 5 到 60。默认为 10。

• ② 每个子任务最多读取的行数。必须大于等于 200000。默认是 200000。

• ③ 每个子任务最多读取的字节数。单位是字节，范围是 100MB 到 1GB。默认是100MB。

这三个参数，用于控制一个子任务的执行时间和处理量。当任意一个达到阈值，则任务结束。 例如：

"max_batch_interval" = "20",
"max_batch_rows" = "300000",
"max_batch_size" = "209715200"

（3）max_error_number

采样窗口内，允许的最大错误行数。必须大于等于 0。默认是 0，即不允许有错误行。

采样窗口为 max_batch_rows * 10。即如果在采样窗口内，错误行数大于 max_error_number，则会导致例行作业被暂停，需要人工介入检查数据质量问题。 被 where 条件过滤掉的行不算错误行

（4）strict_mode

是否开启严格模式，默认为关闭。如果开启后，非空原始数据的列类型变换如果结果为NULL，则会被过滤。指定方式为 "strict_mode" = "true"

（5）timezone

指定导入作业所使用的时区。默认为使用 Session 的 timezone 参数。该参数会影响所有导入涉及的和时区有关的函数结果

（6）format

指定导入数据格式，默认是 csv，支持 json 格式

（7）jsonpaths

jsonpaths: 导入 json 方式分为：简单模式和匹配模式。如果设置了jsonpath 则为匹配模式导入，否则为简单模式导入，具体可参考示例

（8）strip_outer_array

布尔类型，为 true 表示 json 数据以数组对象开始且将数组对象中进行展平，默认值是false

（9）json_root

json_root 为合法的 jsonpath 字符串，用于指定 json document 的根节点，默认值为""

（10）send_batch_parallelism

整型，用于设置发送批处理数据的并行度，如果并行度的值超过BE配置中的max_send_batch_parallelism_per_job，那么作为协调点的BE将使用max_send_batch_parallelism_per_job 的值

6）data_source_properties

数据源的类型。当前支持：Kafka

("key1" = "val1","key2" = "val2"
)