正则表达式匹配网站,如何在局域网做网站,中国电子工程师网,qq电脑版例行导入#xff08;Routine Load#xff09;功能为用户提供了一种自动从指定数据源进行数据导入的功能。
适用场景
当前仅支持从 Kafka 系统进行例行导入#xff0c;使用限制#xff1a;
#xff08;1#xff09;支持无认证的 Kafka 访问#xff0c;以及通过 SSL 方…例行导入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 提交一个例行导入作业。
1FE 通过 JobScheduler 将一个导入作业拆分成若干个 Task。每个 Task 负责导入指定的一部分数据。Task 被 TaskScheduler 分配到指定的 BE 上执行。
2在 BE 上一个 Task 被视为一个普通的导入任务通过 Stream Load 的导入机制进行导入。导入完成后向 FE 汇报。
3FE 中的 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 在运行。
2tbl_name
指定需要导入的表的名称。
3merge_type
数据的合并类型一共支持三种类型 APPEND、DELETE、MERGE 其中APPEND 是默认值表示这批数据全部需要追加到现有数据中DELETE 表示删除与这批数据 key 相同的所有行MERGE 语义 需要与 delete on 条件联合使用表示满足 delete 条件的数据按照 DELETE 语义处理其余的按照 APPEND 语义处理 , 语法为
[WITHMERGE|APPEND|DELETE]4load_properties
用于描述导入数据。语法
[column_separator], [columns_mapping], [where_predicates], [delete_on_predicates], [source_sequence], [partitions], [preceding_predicates]1column_separator:
指定列分隔符如 COLUMNS TERMINATED BY , 这个只在文本数据导入的时候需要指定JSON 格式的数据导入不需要指定这个参数。
默认为\t
2columns_mapping:
指定源数据中列的映射关系以及定义衍生列的生成方式。
映射列
按顺序指定源数据中各个列对应目的表中的哪些列。对于希望跳过的列可以指定一个不存在的列名。假设目的表有三列 k1, k2, v1。源数据有 4 列其中第 1、2、4 列分别对应 k2, k1, v1。则书写如下
COLUMNS (k2, k1, xxx, v1)
其中 xxx 为不存在的一列用于跳过源数据中的第三列。
衍生列
以 col_name expr 的形式表示的列我们称为衍生列。即支持通过 expr 计算得出目的表中对应列的值。 衍生列通常排列在映射列之后虽然这不是强制的规定但是 Doris 总是先解析映射列再解析衍生列。 接上一个示例假设目的表还有第 4 列 v2v2 由 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)3where_predicates
用于指定过滤条件以过滤掉不需要的列。过滤列可以是映射列或衍生列。 例如我们只希望导入 k1 大于 100 并且 k2 等于 1000 的列则书写如下
WHERE k1 100 and k2 10004partitions
指定导入目的表的哪些 partition 中。如果不指定则会自动导入到对应的 partition 中。
示例
PARTITION(p1, p2, p3)5delete_on_predicates
表示删除条件仅在 merge type 为 MERGE 时有意义语法与 where 相同
6source_sequence:
只适用于 UNIQUE_KEYS,相同 key 列下保证 value 列按照 source_sequence 列进行REPLACE, source_sequence 可以是数据源中的列也可以是表结构中的一列。
7preceding_predicates
PRECEDING FILTER predicate用于过滤原始数据。原始数据是未经列映射、转换的数据。用户可以在对转换前的数据前进行一次过滤选取期望的数据再进行转换。
5job_properties
用于指定例行导入作业的通用参数。 语法
PROPERTIES (key1 val1,key2 val2
)目前支持以下参数
1desired_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 节点数。
2max_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 2097152003max_error_number
采样窗口内允许的最大错误行数。必须大于等于 0。默认是 0即不允许有错误行。
采样窗口为 max_batch_rows * 10。即如果在采样窗口内错误行数大于 max_error_number则会导致例行作业被暂停需要人工介入检查数据质量问题。 被 where 条件过滤掉的行不算错误行
4strict_mode
是否开启严格模式默认为关闭。如果开启后非空原始数据的列类型变换如果结果为NULL则会被过滤。指定方式为 strict_mode true
5timezone
指定导入作业所使用的时区。默认为使用 Session 的 timezone 参数。该参数会影响所有导入涉及的和时区有关的函数结果
6format
指定导入数据格式默认是 csv支持 json 格式
7jsonpaths
jsonpaths: 导入 json 方式分为简单模式和匹配模式。如果设置了jsonpath 则为匹配模式导入否则为简单模式导入具体可参考示例
8strip_outer_array
布尔类型为 true 表示 json 数据以数组对象开始且将数组对象中进行展平默认值是false
9json_root
json_root 为合法的 jsonpath 字符串用于指定 json document 的根节点默认值为
10send_batch_parallelism
整型用于设置发送批处理数据的并行度如果并行度的值超过BE配置中的max_send_batch_parallelism_per_job那么作为协调点的BE将使用max_send_batch_parallelism_per_job 的值
6data_source_properties
数据源的类型。当前支持Kafka
(key1 val1,key2 val2
)
