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

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

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

小新 Air-14 2021 Intel处理器ITL版(82FF)原厂Win11系统

链接&#xff1a;https://pan.baidu.com/s/1EkqpdGcixCNER5uP5yIc4Q?pwddm1d 提取码&#xff1a;dm1d lenovo联想小新Air14笔记本2021款【82FF】原装出厂Windows11系统镜像ISO文件 系统自带所有驱动、出厂主题壁纸、系统属性专属LOGO标志、Office办公软件、联想电脑管家等…

手机如何去图片水印?试试这三种方法

手机如何去图片水印&#xff1f;去水印已然成为了自媒体从业者必备技能之一&#xff0c;无论是工作或生活中经常遇到图片/视频上带有水印&#xff0c;非常影响整体观感&#xff0c;网上去水印方法又很多&#xff0c;如果你是小白&#xff0c;这篇文章将提供给你三个实用去水印的…

草图大师sketchup道路怎么快速种树?

草图大师sketchup道路怎么快速种树&#xff1f;草图大师中的道路图纸想要在道路两旁种树&#xff0c;该怎么快速给道路种树呢&#xff1f;下面我们就来看看详细的教程&#xff0c;需要的朋友可以参考下 草图大师sketchup中想要快速种树&#xff0c;该怎么种多棵树呢&#xff1…

数据结构校招知识点总结

文章目录 前言1. 数据结构概论、算法设计与分析1.1 数据结构三要素&#xff1f;1.2 算法的基本概念&#xff1f;1.3 什么是时间复杂度&#xff1f; 2. 线性表2.1 链表结构和顺序存储结构的区别&#xff1f;2.2 单链表和双链表的区别&#xff1f;2.3 头指针和头结点的区别&#…

基于Python的网络爬虫设计与实现

基于Python的网络爬虫设计与实现 摘要&#xff1a;从互联网时代开始&#xff0c;网络搜索引擎就变得越发重要。大数据时代&#xff0c;一般的网络搜索引擎不能满足用户的具体需求&#xff0c;人们更加注重特定信息的搜索效率&#xff0c;网络爬虫技术应运而生。本设计先对指定…

上海数交所与合合信息发布产业数据行业创新中心,政产学研合力为“数据航母”加速

大数据产业是数字经济创新发展、加速发展的重要方向。11月25日&#xff0c;2023全球数商大会在上海盛大开幕。大会以“数联全球、商通未来”为主题&#xff0c;聚焦数字经济时代下&#xff0c;数据要素推动实体经济发展的规划与成果&#xff0c;是数据交易领域的行业级峰会和数…

Kafka 如何保证消息消费的全局顺序性

哈喽大家好&#xff0c;我是咸鱼 今天我们继续来讲一讲 Kafka 当有消息被生产出来的时候&#xff0c;如果没有指定分区或者指定 key &#xff0c;那么消费会按照【轮询】的方式均匀地分配到所有可用分区中&#xff0c;但不一定按照分区顺序来分配 我们知道&#xff0c;在 Kaf…

动态:class和:style绑定

1. 在应用界面中, 某个(些)元素的样式是变化的 class/style绑定就是专门用来实现动态样式效果的技术 2. 动态class绑定 :class等号后的变量值 可以是字符串 :class等号后 可以是对象 :class等号后 可以是数组 3. 动态style绑定 :style"{ color: myPinkColor, fontS…

【软件测试】银行测试项目VS常规项目有什么区别?怎么做?

目录&#xff1a;导读 前言一、Python编程入门到精通二、接口自动化项目实战三、Web自动化项目实战四、App自动化项目实战五、一线大厂简历六、测试开发DevOps体系七、常用自动化测试工具八、JMeter性能测试九、总结&#xff08;尾部小惊喜&#xff09; 前言 银行测试项目跟常…

ESP32-Web-Server编程- JS 基础5

ESP32-Web-Server编程- JS 基础5 概述 JS 编程内容颇多&#xff0c;我们提供一些简单的示例&#xff0c;先玩再学&#xff0c;边玩边学。 示例1-演示通过 JS 进行温度转换 资源链接 对应示例的 code 链接 &#xff08;点击直达代码仓库&#xff09; 示例2-增加网页弹窗 演…

【攻防世界-misc】reverseMe

1.下载后&#xff0c;得到这样一张图片 2.利用在线翻转网站获取值&#xff0c;在线旋转图片工具|在线翻转照片|调整照片方向|生成镜像图片 - 改图宝 反转后的图片&#xff0c;将值提取并上传。

Java零基础——Nginx篇

1.【熟悉】服务器概述 1.1 目前常见的web服务器 1&#xff0c;Apache(http://httpd.apache.org) 它是世界上用的最多的web服务器&#xff0c;市场占有率达60%左右&#xff0c;模块非常丰富&#xff0c;系统非常稳定&#xff0c;可移植性好&#xff0c;但是比较消耗资源 2&…

Rust之构建命令行程序(一):接受命令行参数

开发环境 Windows 10Rust 1.73.0 VS Code 1.84.2 项目工程 这次创建了新的工程minigrep. IO工程&#xff1a;构建命令行程序 这一章回顾了到目前为止你所学的许多技能&#xff0c;并探索了一些更标准的库特性。我们将构建一个与文件和命令行输入/输出交互的命令行工具&#…

业务流程图用什么软件绘制?

在企业的日常工作中&#xff0c;对于业务流程的把控和优化显得非常重要。为了更好地理解和管理业务流程&#xff0c;业务流程图便应运而生。 业务流程图是企业管理的图形化工具&#xff0c;它描述了企业在生产和服务提供过程中&#xff0c;在各个环节中所涉及的各种操作、任务…

python如何抓取携程酒店的价格,让工作更简单点

有时候老板没事安排点事&#xff0c;为了偷懒&#xff0c;只能使出大招&#xff0c;毕竟自己不是那么老老实实干活的人&#xff0c;整理数据这类累和繁琐的活&#xff0c;我怎么能轻易动&#xff0c;好在gpt可以帮我来实现&#xff0c;有人可能会说&#xff0c;这么点内容你还不…

OpenCvSharp从入门到实践-(05)通道

目录 1、拆分通道 1.1、实例1-拆分一副BGR图像的通道 1.2、实例2-拆分一副HSV图像的通道 2、合并通道 2.1 实例3-合并通道的顺序不同&#xff0c;图像的显示效果也不通 实例4-合格H通道图像、S通道图像、V通道图像 3、综合运用拆分通道和合并通道 2.2、实例5-只把H通道的…

Vue3-admin-template 导入模板功能

先看效果&#xff1a; 直接上代码&#xff1a; 1.绑定事件&#xff1a; <el-button type"primary" click"templates">模板导入</el-button> 2.写结构样式 <!-- 模板导入 --><el-dialog v-model"Statusimprot" title&quo…

js相同字符串截取拼接

原数据 const list [999-1234567801,999-1234567802,999-1234567803, ]; const list1 [999-1234567899,999-1234567900,999-1234567901, ];期望数据 999-1234567801/2/3 //list 999-1234567899/900/901 //list1处理代码 // 连续号码处理 export const formatNumber (tick…

9.二维数组——打印出杨辉三角形(要求打印出10行)

文章目录 前言一、题目描述 二、题目分析 三、解题 程序运行代码 前言 本系列为二维数组编程题&#xff0c;点滴成长&#xff0c;一起逆袭。 一、题目描述 打印出杨辉三角形&#xff08;要求打印出10行&#xff09;。 二、题目分析 三、解题 程序运行代码 #include<s…

在柯桥西班牙语论文写作,连接词只会用porquepero?西语连接词大全来啦~

Adicin -agregan nuevos datos al desarrollo de una idea o introduce otro aspecto del tema. 为观点的论述增添新的信息&#xff0c;或介绍主题的另一个方面。 1 Lista de conectores 连接词列表 Adems. As mismo. Hay que mencionar, adems. Habra que decir tambin. Mas …