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"
)

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:/a/195605.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

解析直播第三方美颜SDK:技术原理与应用

时下&#xff0c;直播平台和主播们纷纷引入美颜技术&#xff0c;以提升视觉效果和用户体验。而在众多美颜技术中&#xff0c;直播第三方美颜SDK成为许多开发者和平台的首选&#xff0c;因其灵活性和高效性而备受推崇。 一、技术原理&#xff1a;美颜算法的精髓 第三方美颜SDK…

02-Java集合之双列集合,如HashMap,Hashtable,Properties,TreeMap的底层结构

双列集合 添加/获取/删除键值对原理 哈希表/散列表是一种将数组和单向链表融合在一起的数据结构 数组在查询方面效率很高,单向链表在随机增删方面效率较高,哈希表将以上的两种数据结构融合在一起后充分发挥它们各自的优点 双列集合以key和value这种键值对方式存储数据: key…

ViLT 论文精读【论文精读】

ViLT 论文精读【论文精读】_哔哩哔哩_bilibili 目录 ViLT 论文精读【论文精读】_哔哩哔哩_bilibili 1 地位 2 ViLT做了什么能让它成为这种里程碑式的工作&#xff1f; 3 ViLT到底把模型简化到了什么程度&#xff1f;到底能加速到什么程度&#xff1f; 2.1 过去的方法是怎…

bop数据合并到COCO

bop数据合并到COCO JSON转TXT重命名txt文件中类别信息的转换 JSON转TXT import json import os,globcategories [{"id": 12,"name": "OREO","supercategory": "icbin"},{"id": 16,"name": "Paper…

Rust语言入门教程(七) - 所有权系统

所有权系统是Rust敢于声称自己为一门内存安全语言的底气来源&#xff0c;也是让Rust成为一门与众不同的语言的所在之处。也正是因为这个特别的所有权系统&#xff0c;才使得编译器能够提前暴露代码中的错误&#xff0c;并给出我们必要且精准的错误提示。 所有权系统的三个规则…

微信小程序生成二维码并保存到本地方法

微信小程序生成二维码请保存到本地方法 官方weapp-qrcode插件 github链接 功能完成样子 wxml <view class"qrcode"><canvas style"width: 275px; height: 275px;" canvas-idmyQrcode></canvas> </view> <view class" …

Transformers实战——文本相似度

文章目录 一、改写文本分类1.导入相关包2.加载数据集3.划分数据集4.数据集预处理5.创建模型6.创建评估函数7.创建 TrainingArguments8.创建 Trainer9.模型训练10.模型评估11.模型预测 二、交互/单塔模式1.导入相关包2.加载数据集3.划分数据集4.数据集预处理5.创建模型&#xff…

机器学习的复习笔记2-回归

一、什么是回归 机器学习中的回归是一种预测性分析任务&#xff0c;旨在找出因变量&#xff08;目标变量&#xff09;和自变量&#xff08;预测变量&#xff09;之间的关系。与分类问题不同&#xff0c;回归问题关注的是预测连续型或数值型数据&#xff0c;如温度、年龄、薪水…

springboot自定义更换启动banner动画

springboot自定义更换启动banner动画 文章目录 springboot自定义更换启动banner动画 &#x1f4d5;1.新建banner&#x1f5a5;️2.启动项目&#x1f516;3.自动生成工具&#x1f9e3;4.彩蛋 &#x1f58a;️最后总结 &#x1f4d5;1.新建banner 在resources中新建banner.txt文…

数智赋能 锦江汽车携手苏州金龙打造高质量盛会服务

作为一家老牌客运公司&#xff0c;成立于1956年的上海锦江汽车服务有限公司&#xff08;以下简称锦江汽车&#xff09;&#xff0c;拥有1200多辆大巴和5000多辆轿车&#xff0c;是上海乃至长三角地区规模最大的专业旅游客运公司。面对客运市场的持续萎缩&#xff0c;锦江汽车坚…

打破TikTok信息壁垒:东南亚达人不实名就封小黄车?跨境新店考核规则更改!

近期&#xff0c;TKFFF得到一个模糊消息&#xff0c;不保真但是个合规的趋势&#xff0c;分享给大家&#xff01; 目前泰国已落实达人实名制&#xff0c;11月底或者12月底之前可能东南亚所有达人都需要实名制&#xff0c;后期也会同步到全球。 如果达人不实名&#xff0c;会被…

蓝桥杯-平方和(599)

【题目】平方和 【通过测试】代码 import java.util.Scanner; import java.util.ArrayList; import java.util.List; // 1:无需package // 2: 类名必须Main, 不可修改public class Main {public static void main(String[] args) {Scanner scan new Scanner(System.in);//在此…

有哪些可信的SSL证书颁发机构?

目前市面上所显示的SSL证书颁发机构可所谓不计其数&#xff0c;类型也是多样&#xff0c;就好比我们同样是买一件T恤&#xff0c;却有百家不同类型的店铺一个道理。根据CA里面看似很多&#xff0c;但能拿到99%浏览器及设备信任度的寥寥无几&#xff0c;下面小编整理出几家靠谱可…

【substance painter】如何制作一个生锈磨损的枪

&#x1f468;‍&#x1f4bb;个人主页&#xff1a;元宇宙-秩沅 &#x1f468;‍&#x1f4bb; hallo 欢迎 点赞&#x1f44d; 收藏⭐ 留言&#x1f4dd; 加关注✅! &#x1f468;‍&#x1f4bb; 本文由 秩沅 原创 &#x1f468;‍&#x1f4bb; 收录于专栏&#xff1a;Uni…

提升企业文档处理效率,尽在Readiris PDF Corporate下载

在现代企业中&#xff0c;大量的文档处理工作是不可避免的。然而&#xff0c;传统的文档处理方法往往效率低下&#xff0c;浪费了企业宝贵的时间和资源。为了帮助企业提升文档处理效率&#xff0c;我们推荐使用Readiris PDF Corporate软件。 Readiris PDF Corporate是一款功能…

UE4/UE5 雾

雾 UE4/UE5 雾平面雾材质效果图&#xff1a; 3D雾区材质效果图&#xff1a; UE4/UE5 雾 平面雾 做好将材质放在Plane上就行 材质 效果图&#xff1a; 3D雾区 做好将材质放在3D模型上就行 材质 效果图&#xff1a;

反转链表的三种写法

题目链接&#xff1a;https://leetcode.cn/problems/reverse-linked-list/ 方法一&#xff1a;循环&#xff0c;维护好两个节点一个前一个后 class Solution {public ListNode reverseList(ListNode head) {ListNode pre null;ListNode local head;while(local ! null){List…

信奥编程 1168:大整数加法

解析&#xff1a;在c中需要考虑这么几个问题&#xff0c;第一个是大数据的输入&#xff0c;第二个是大数据的存储&#xff0c;第三是大数据的计算方式&#xff0c;最后是输出。 针对上述几个问题&#xff0c;第一个问题&#xff0c;采用字符串的方式或者数组加循环的方式接收输…

Typora+PicGo+Minio搭建博客图床

文章目录 TyporaPicGoMinio搭建博客图床前言什么是图床?为什么需要图床?准备工作一、Typora二、Picgo1. 下载Picgo2. 下载node.js3. 下载minio插件 三、服务器端配置1. 添加端口到安全组2. 使用Docker安装minio3. 配置minio image-20231127175530696四、minio插件配置五、Typ…

c++_继承

&#x1f3f7;如被何实现一个不能被继承的类&#xff08;或是继承无意义的类&#xff09; 将构造函数定义成私有的就行了&#xff0c;即&#xff1a;私有化父类的构造函数 c 11 新增关键字final 修饰父类直接不能被继承 class A final {........ }&#x1f3f7;继承与有元 有…