Flink 如何读取和写入 Clickhouse?
liuian 2025-01-08 15:17 21 浏览
Flink 从入门到精通 系列文章
?问题导读:
1.Flink读写Clickhouse支持哪个版本?
2.ClickHouse读写Clickhouse有哪些参数?
3.ClickHouse读写Clickhouse参数都有哪些说明?
?
ClickHouse 读取
一、插件名称
名称:clickhousereader
二、支持的数据源版本
ClickHouse 19.x及以上
三、参数说明
「jdbcUrl」
描述:针对关系型数据库的jdbc连接字符串
jdbcUrl参考文档:clickhouse-jdbc官方文档
必选:是
默认值:无
「username」
描述:数据源的用户名
必选:是
默认值:无
「password」
描述:数据源指定用户名的密码
必选:是
默认值:无
「where」
描述:筛选条件,reader插件根据指定的column、table、where条件拼接SQL,并根据这个SQL进行数据抽取。在实际业务场景中,往往会选择当天的数据进行同步,可以将where条件指定为gmt_create > time。
注意:不可以将where条件指定为limit 10,limit不是SQL的合法where子句。
必选:否
默认值:无
「splitPk」
描述:当speed配置中的channel大于1时指定此参数,Reader插件根据并发数和此参数指定的字段拼接sql,使每个并发读取不同的数据,提升读取速率。注意:推荐splitPk使用表主键,因为表主键通常情况下比较均匀,因此切分出来的分片也不容易出现数据热点。目前splitPk仅支持整形数据切分,不支持浮点、字符串、日期等其他类型。如果用户指定其他非支持类型,FlinkX将报错!如果channel大于1但是没有配置此参数,任务将置为失败。
必选:否
默认值:无
「fetchSize」
描述:读取时每批次读取的数据条数。
注意:此参数的值不可设置过大,否则会读取超时,导致任务失败。
必选:否
默认值:1000
「queryTimeOut」
描述:查询超时时间,单位秒。
注意:当数据量很大,或者从视图查询,或者自定义sql查询时,可通过此参数指定超时时间。
必选:否
默认值:1000
「customSql」
描述:自定义的查询语句,如果只指定字段不能满足需求时,可通过此参数指定查询的sql,可以是任意复杂的查询语句。注意:只能是查询语句,否则会导致任务失败;查询语句返回的字段需要和column列表里的字段严格对应;当指定了此参数时,connection里指定的table无效;当指定此参数时,column必须指定具体字段信息,不能以*号代替;
必选:否
默认值:无
「column」
描述:需要读取的字段。
格式:支持3种格式
1.读取全部字段,如果字段数量很多,可以使用下面的写法:
"column":["*"]
2.只指定字段名称:
"column":["id","name"]
3.指定具体信息:
"column": [{
"name": "col",
"type": "datetime",
"format": "yyyy-MM-dd hh:mm:ss",
"value": "value"
}]
属性说明:
?name:字段名称
type:字段类型,可以和数据库里的字段类型不一样,程序会做一次类型转换
format:如果字段是时间字符串,可以指定时间的格式,将字段类型转为日期格式返回
value:如果数据库里不存在指定的字段,则会报错。如果指定的字段存在,当指定字段的值为时,会以此value值作为默认值返回
?
必选:是
默认值:无
「polling」
描述:是否开启间隔轮询,开启后会根据pollingInterval轮询间隔时间周期性的从数据库拉取数据。开启间隔轮询还需配置参数pollingInterval,increColumn,可以选择配置参数startLocation。若不配置参数startLocation,任务启动时将会从数据库中查询增量字段最大值作为轮询的开始位置。
必选:否
默认值:false
「pollingInterval」
描述:轮询间隔时间,从数据库中拉取数据的间隔时间,默认为5000毫秒。
必选:否
默认值:5000
「requestAccumulatorInterval」
描述:发送查询累加器请求的间隔时间。
必选:否
默认值:2
配置示例
1、基础配置
{
"job": {
"content": [{
"reader": {
"parameter" : {
"column" : [ {
"name" : "id",
"type" : "bigint",
"key" : "id"
}, {
"name" : "user_id",
"type" : "bigint",
"key" : "user_id"
}, {
"name" : "name",
"type" : "varchar",
"key" : "name"
} ],
"username" : "username",
"password" : "password",
"connection" : [ {
"jdbcUrl" : [ "jdbc:clickhouse://0.0.0.1:8123/dtstack" ],
"table" : [ "tableTest" ]
} ],
"where": "id > 1",
"splitPk": "id",
"fetchSize": 1000,
"queryTimeOut": 1000,
"customSql": "",
"requestAccumulatorInterval": 2
},
"name" : "clickhousereader"
},
"writer": {
"name": "streamwriter",
"parameter": {
"print": true
}
}
}],
"setting": {
"speed": {
"channel": 1,
"bytes": 0
},
"errorLimit": {
"record": 100
}
}
}
}
2、多通道
{
"job": {
"content": [{
"reader": {
"parameter" : {
"column" : [ {
"name" : "id",
"type" : "bigint",
"key" : "id"
}, {
"name" : "user_id",
"type" : "bigint",
"key" : "user_id"
}, {
"name" : "name",
"type" : "varchar",
"key" : "name"
} ],
"username" : "username",
"password" : "password",
"connection" : [ {
"jdbcUrl" : [ "jdbc:clickhouse://0.0.0.1:8123/dtstack" ],
"table" : [ "tableTest" ]
} ],
"where": "id > 1",
"splitPk": "id",
"fetchSize": 1000,
"queryTimeOut": 1000,
"customSql": "",
"requestAccumulatorInterval": 2
},
"name" : "clickhousereader"
},
"writer": {
"name": "streamwriter",
"parameter": {
"print": true
}
}
}],
"setting": {
"speed": {
"channel": 3,
"bytes": 0
},
"errorLimit": {
"record": 100
}
}
}
}
3、指定customSql
{
"job": {
"content": [{
"reader": {
"parameter" : {
"column" : [ {
"name" : "id",
"type" : "bigint",
"key" : "id"
}, {
"name" : "user_id",
"type" : "bigint",
"key" : "user_id"
}, {
"name" : "name",
"type" : "varchar",
"key" : "name"
} ],
"username" : "username",
"password" : "password",
"connection" : [ {
"jdbcUrl" : [ "jdbc:clickhouse://0.0.0.1:8123/dtstack" ],
"table" : [ "tableTest" ]
} ],
"where": "id > 1",
"splitPk": "id",
"fetchSize": 1000,
"queryTimeOut": 1000,
"customSql": "select id from tableTest",
"requestAccumulatorInterval": 2
},
"name" : "clickhousereader"
},
"writer": {
"name": "streamwriter",
"parameter": {
"print": true
}
}
}],
"setting": {
"speed": {
"channel": 1,
"bytes": 0
},
"errorLimit": {
"record": 100
}
}
}
}
4、增量同步指定startLocation
{
"job": {
"content": [{
"reader": {
"parameter" : {
"column" : [ {
"name" : "id",
"type" : "bigint",
"key" : "id"
}, {
"name" : "user_id",
"type" : "bigint",
"key" : "user_id"
}, {
"name" : "name",
"type" : "varchar",
"key" : "name"
} ],
"username" : "username",
"password" : "password",
"connection" : [ {
"jdbcUrl" : [ "jdbc:clickhouse://0.0.0.1:8123/dtstack" ],
"table" : [ "tableTest" ]
} ],
"where": "id > 1",
"splitPk": "id",
"fetchSize": 1000,
"queryTimeOut": 1000,
"customSql": "",
"increColumn": "id",
"startLocation": "20",
"requestAccumulatorInterval": 2
},
"name" : "clickhousereader"
},
"writer": {
"name": "streamwriter",
"parameter": {
"print": true
}
}
}],
"setting": {
"speed": {
"channel": 1,
"bytes": 0
},
"errorLimit": {
"record": 100
}
}
}
}
5、间隔轮询
{
"job": {
"content": [{
"reader": {
"parameter" : {
"column" : [ {
"name" : "id",
"type" : "bigint",
"key" : "id"
}, {
"name" : "user_id",
"type" : "bigint",
"key" : "user_id"
}, {
"name" : "name",
"type" : "varchar",
"key" : "name"
} ],
"username" : "username",
"password" : "password",
"connection" : [ {
"jdbcUrl" : [ "jdbc:clickhouse://0.0.0.1:8123/dtstack" ],
"table" : [ "tableTest" ]
} ],
"where": "id > 1",
"splitPk": "id",
"fetchSize": 1000,
"queryTimeOut": 1000,
"customSql": "",
"requestAccumulatorInterval": 2,
"polling": true,
"pollingInterval": 3000
},
"name" : "clickhousereader"
},
"writer": {
"name": "streamwriter",
"parameter": {
"print": true
}
}
}],
"setting": {
"speed": {
"channel": 1,
"bytes": 0
},
"errorLimit": {
"record": 100
}
}
}
}
ClickHouse 写入
一、插件名称
名称:clickhousewriter
二、支持的数据源版本
ClickHouse 19.x及以上
三、参数说明
「jdbcUrl」
描述:针对关系型数据库的jdbc连接字符串
必选:是
默认值:无
「username」
描述:数据源的用户名
必选:是
默认值:无
「password」
描述:数据源指定用户名的密码
必选:是
默认值:无
「column」
描述:目的表需要写入数据的字段,字段之间用英文逗号分隔。例如: "column": ["id","name","age"]。
必选:是
默认值:否
默认值:无
「preSql」
描述:写入数据到目的表前,会先执行这里的一组标准语句
必选:否
默认值:无
「postSql」
描述:写入数据到目的表后,会执行这里的一组标准语句
必选:否
默认值:无
「table」
描述:目的表的表名称。目前只支持配置单个表,后续会支持多表
必选:是
默认值:无
「writeMode」
描述:控制写入数据到目标表采用 insert into 语句,只支持insert操作
必选:是
所有选项:insert
默认值:insert
「batchSize」
描述:一次性批量提交的记录数大小,该值可以极大减少FlinkX与数据库的网络交互次数,并提升整体吞吐量。但是该值设置过大可能会造成FlinkX运行进程OOM情况
必选:否
默认值:1024
「插件下载」
公众号内回复 「flinkx」即可获取
?转载:https://www.aboutyun.com/forum.php?mod=viewthread&tid=29271
?
基于 Apache Flink 的实时监控告警系统
关于数据中台的深度思考与总结(干干货)
日志收集Agent,阴暗潮湿的地底世界
2020 继续踏踏实实的做好自己
你点的每个赞,我都当成了喜欢
相关推荐
- vue怎么和后端php配合
-
Vue和后端PHP可以通过HTTP请求进行配合。首先,前端Vue可以使用axios库或者Vue自带的$http对象来发送HTTP请求到后端PHP接口。通过axios库发送POST、GET、PUT等请求...
- Ansible最佳实践之 AWX 使用 Ansible 与 API 通信
-
#头条创作挑战赛#API简单介绍红帽AWX提供了一个类似Swagger的RESTful风格的Web服务框架,可以和awx直接交互。使管理员和开发人员能够在webUI之外控制其...
- PHP8.3 错误处理革命:Exception 与 Error 全面升级
-
亲爱的小伙伴,好久没有发布信息了,最近学习了一下PHP8.3的升级,都有哪些优化和提升,把学到的分享出来给需要的小伙伴充下电。技术段位:高可用性必修目标收益:精准错误定位+异常链路追踪适配场景...
- 使用 mix/vega + mix/db 进行现代化的原生 PHP 开发
-
最近几年在javascript、golang生态中游走,发现很多npm、gomod的优点。最近回过头开发MixPHPV3,发现composer其实一直都是一个非常优秀的工具,但是...
- 15 个非常好用的 JSON 工具
-
JSON(JavaScriptObjectNotation)是一种流行的数据交换格式,已经成为许多应用程序中常用的标准。无论您是开发Web应用程序,构建API,还是处理数据,使用JSON工具可以大...
- php8环境原生实现rpc
-
大数据分布式架构盛行时代的程序员面试,常常遇到分布式架构,RPC,本文的主角是RPC,英文名为RemoteProcedureCall,翻译过来为“远程过程调用”。主流的平台中都支持各种远程调用技术...
- 「PHP编程」如何搭建私有Composer包仓库?
-
在前一篇文章「PHP编程」如何制作自己的Composer包?中,我们已经介绍了如何制作自己的composer包,以及如何使用composer安装自己制作的composer包。不过,这其中有...
- WAF-Bypass之SQL注入绕过思路总结
-
过WAF(针对云WAF)寻找真实IP(源站)绕过如果流量都没有经过WAF,WAF当然无法拦截攻击请求。当前多数云WAF架构,例如百度云加速、阿里云盾等,通过更改DNS解析,把流量引入WAF集群,流量经...
- 【推荐】一款 IDEA 必备的 JSON 处理工具插件 — Json Assistant
-
JsonAssistant是基于IntelliJIDEs的JSON工具插件,让JSON处理变得更轻松!主要功能完全支持JSON5JSON窗口(多选项卡)选项卡更名移动至主编辑器用...
- 技术分享 | 利用PHAR协议进行PHP反序列化攻击
-
PHAR(“PhpARchive”)是PHP中的打包文件,相当于Java中的JAR文件,在php5.3或者更高的版本中默认开启。PHAR文件缺省状态是只读的,当我们要创建一个Phar文件需要修改...
- php进阶到架构之swoole系列教程(一)windows安装swoole
-
目录概述安装Cygwin安装swoolephp7进阶到架构师相关阅读概述这是关于php进阶到架构之swoole系列学习课程:第一节:windows安装swoole学习目标:在Windows环境将搭建s...
- go 和 php 性能如何进行对比?
-
PHP性能很差吗?每次讲到PHP和其他语言间的性能对比,似乎都会发现这样一个声音:单纯的性能对比没有意义,主要瓶颈首先是数据库,其次是业务代码等等。好像PHP的性能真的不能单独拿出来讨论似的。但其实一...
- Linux(CentOS )手动搭建LNMP(Linux+Nginx+Mysql+PHP)坏境
-
CentOS搭建LNMP(Linux+Nginx+Mysql+PHP)坏境由于网上各种版本新旧不一,而且Linux版本也不尽相同,所以自己写一遍根据官网的提示自己手动搭建过程。看官方文档很重要,永远...
- json和jsonp区别
-
JSON和JSONP虽然只有一个字母的差别,但其实他们根本不是一回事儿:JSON是一种数据交换格式,而JSONP是一种非官方跨域数据交互协议。一个是描述信息的格式,一个是信息传递的约定方法。一、...
- web后端正确的返回JSON
-
在web开发中,前端和后端发生数据交换传输现在最常见的形式就是异步ajax交互,一般返回给js都是json,如何才是正确的返回呢?前端代码想要获取JSON数据代码如下:$.get('/user-inf...
- 一周热门
-
-
Python实现人事自动打卡,再也不会被批评
-
Psutil + Flask + Pyecharts + Bootstrap 开发动态可视化系统监控
-
一个解决支持HTML/CSS/JS网页转PDF(高质量)的终极解决方案
-
【验证码逆向专栏】vaptcha 手势验证码逆向分析
-
再见Swagger UI 国人开源了一款超好用的 API 文档生成框架,真香
-
网页转成pdf文件的经验分享 网页转成pdf文件的经验分享怎么弄
-
C++ std::vector 简介
-
python使用fitz模块提取pdf中的图片
-
《人人译客》如何规划你的移动电商网站(2)
-
Jupyterhub安装教程 jupyter怎么安装包
-
- 最近发表
- 标签列表
-
- python判断字典是否为空 (50)
- crontab每周一执行 (48)
- aes和des区别 (43)
- bash脚本和shell脚本的区别 (35)
- canvas库 (33)
- dataframe筛选满足条件的行 (35)
- gitlab日志 (33)
- lua xpcall (36)
- blob转json (33)
- python判断是否在列表中 (34)
- python html转pdf (36)
- 安装指定版本npm (37)
- idea搜索jar包内容 (33)
- css鼠标悬停出现隐藏的文字 (34)
- linux nacos启动命令 (33)
- gitlab 日志 (36)
- adb pull (37)
- table.render (33)
- uniapp textarea (33)
- python判断元素在不在列表里 (34)
- python 字典删除元素 (34)
- react-admin (33)
- vscode切换git分支 (35)
- vscode美化代码 (33)
- python bytes转16进制 (35)