当前位置: 首页 > news >正文

【自定义Source、Sink】Flink自定义Source、Sink对ClickHouse进行读和批量写操作

ClickHouse官网文档

Flink 读取 ClickHouse 数据两种驱动

  1. ClickHouse 官方提供Clickhouse JDBC.【建议使用
  2. 第3方提供的Clickhouse JDBC. ru.yandex.clickhouse.ClickHouseDriver

ru.yandex.clickhouse.ClickHouseDriver.现在是没有维护

ClickHouse 官方提供Clickhouse JDBC的包名:com.clickhouse.jdbc.*

有些版本com.clickhouse.jdbc.* 包含了 ru.yandex.clickhouse.ClickHouseDriver.

因此加载包的时候一定要注意导入的包名

引入依赖

        <!-- clickhouse jdbc driver --><dependency><groupId>com.clickhouse</groupId><artifactId>clickhouse-jdbc</artifactId></dependency>

使用的是 0.3 这个版本,该版本就包含上述3方CH jdbc包

     <!-- CH JDBC版本推荐使用 0.3, 0.4的版本是要 JDK 17 --><clickhouse-jdbc.version>0.3.2-patch11</clickhouse-jdbc.version>

自定义Source

测试表映射实体类,该表仅有一个name字段

@Data
@NoArgsConstructor
@AllArgsConstructor
public class CHTestPO {private String name;}

Flink Clickhouse Source

public class ClickHouseSource implements SourceFunction<CHTestPO> {private final String URL;private final String SQL;public ClickHouseSource(String URL, String SQL) {this.URL = URL;this.SQL = SQL;}@Overridepublic void run(SourceContext<CHTestPO> output) throws Exception {//  Properties是持久化的属性集 Properties的key和value都是字符串Properties properties = new Properties();ClickHouseDataSource clickHouseDataSource = new ClickHouseDataSource(URL, properties);// 使用 try-with-resource 方式关闭JDBC连接 无需手动关闭try (ClickHouseConnection conn = clickHouseDataSource.getConnection()) {// clickhouse 通过游标的方式读取数据Statement stmt = conn.createStatement();ResultSet rs = stmt.executeQuery(SQL);while (rs.next()) {String name = rs.getString(1);output.collect(new CHTestPO(name));}}}@Overridepublic void cancel() {}
}

自定义Sink

需额外引入依赖

        <!-- Flink-Connector-Jdbc --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId></dependency>

Java 对sql语句处理的两个对象

  1. PreparedStatement对象:能够对预编译之后的sql语句进行处理【SQL 语句预编译:通过占位符'?'实现,可以防止sql注入】
  2. Statement对象:只能对静态的sql语句进行处理

核心代码

/*** 使用 Flink-jdbc-connector + 批量写入 + sql语句的预编译 写入 Clickhouse*/
public class ClickHouseJdbcSink<T> {private final SinkFunction<T> sink;private final static String NA = "null";public ClickHouseJdbcSink(String sql, int batchSize, String url) {sink = JdbcSink.sink(sql,// 对sql语句进行预编译new ClickHouseJdbcStatementBuilder<T>(),// 设置批量插入数据new JdbcExecutionOptions.Builder().withBatchSize(batchSize).build(),// 设置ClickHouse连接配置new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl(url).build());}public SinkFunction<T> getSink() {return this.sink;}/*** 对预编译之后的sql语句进行占位符替换** @param ps:     PreparedStatement对象 下标从 1 开始* @param fields: clickhouse表PO对象的属性字段* @param object: clickhouse表PO对象的属性字段所对应的数据类型*/public static void setPreparedStatement(PreparedStatement ps,Field[] fields,Object object) throws IllegalAccessException, SQLException {// 遍历 Field[]for (int i = 1; i <= fields.length; i++) {// 取出每个Field实例Field field = fields[i - 1];// 指示反射的对象在使用时应该取消 Java 语言访问检查field.setAccessible(true);// 通过Field实例的get方法返回指定的对象Object o = field.get(object);if (o == null) {ps.setNull(i, 0);continue;}// 这里统一设为字符型String fieldValue = o.toString();// 变量和常量的比较,通常将常量放前,可以避免空指针if (!NA.equals(fieldValue) && !"".equals(fieldValue)) {// 替换对应位置的占位符ps.setObject(i, fieldValue);} else {ps.setNull(i, 0);}}}}

对sql语句进行预编译

@Slf4j
public class ClickHouseJdbcStatementBuilder<T> implements JdbcStatementBuilder<T> {@Overridepublic void accept(PreparedStatement preparedStatement, T t) throws SQLException {/* *********************** Java通过反射获取类的字段:** 1. getDeclaredFields():获取所有的字段,不会获取父类的字段* 2. getFields(): 只能会public字段,获取包含父类的字段** *********************/Field[] fields = t.getClass().getDeclaredFields();// 将获取到的字段替换sql预编译之后的占位符。try {ClickHouseJdbcSink.setPreparedStatement(preparedStatement, fields, t);} catch (IllegalAccessException e) {log.error("sql 预编译失败", e);e.printStackTrace();}}
}

ClickHouse读写工具类

image-20231209233006017

public class ClickHouseUtil {private static final String URL;static {ParameterTool parameterTool = ParameterUtil.getParameters();URL = parameterTool.get("clickhouse.url");}/*** 读取clickhouse*/public static DataStream<CHTestPO> read(StreamExecutionEnvironment env, String sql) {return env.addSource(new ClickHouseSource(URL, sql));}/*** 批量写入ClickHouse*/public static <T> DataStreamSink<T> batchWrite(DataStream<T> dataStream,String sql,int batchSize) {//生成 SinkFunctionClickHouseJdbcSink<T> clickHouseJdbcSink =new ClickHouseJdbcSink<T>(sql, batchSize, URL);return dataStream.addSink(clickHouseJdbcSink.getSink());}}

测试一下

public class ClickHouseUtilTest {@DisplayName("测试Flink+jdbc+游标读取Clickhouse")@Testvoid testRead() throws Exception {StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();// 设置并行度1env.setParallelism(1);// 从default数据库的user表中读取数据String sql = "select * from default.user";DataStream<CHTestPO> ds = ClickHouseUtil.read(env, sql);// 打印数据流中的元素ds.print("clickhouse");// 执行程序env.execute();}@DisplayName("测试Flink-Connector-jdbc+预编译批量写入Clickhouse")@Testvoid testBatchWrite() throws Exception {StreamExecutionEnvironment env =StreamExecutionEnvironment.getExecutionEnvironment();// 设置并行度1env.setParallelism(1);CHTestPO po = new CHTestPO();po.setName("Lucy");CHTestPO po1 = new CHTestPO();po1.setName("Jack");DataStream<CHTestPO> ds = env.fromCollection(Arrays.asList(po, po1));// 定义将数据写入ClickHouse数据库的SQL语句String sql = "insert into default.user(name) values(?)";// 调用ClickHouseUtil的batchWrite方法将数据流ds中的数据批量写入ClickHouse数据库ClickHouseUtil.batchWrite(ds, sql, 2);// 执行程序env.execute();}
}

此时表中仅一行记录

image-20231209232619959

读取没有问题!

image-20231209232741522

写入没有问题!

image-20231209232902469

相关文章:

【自定义Source、Sink】Flink自定义Source、Sink对ClickHouse进行读和批量写操作

ClickHouse官网文档 Flink 读取 ClickHouse 数据两种驱动 ClickHouse 官方提供Clickhouse JDBC.【建议使用】第3方提供的Clickhouse JDBC. ru.yandex.clickhouse.ClickHouseDriver ru.yandex.clickhouse.ClickHouseDriver.现在是没有维护 ClickHouse 官方提供Clickhouse JDBC…...

linux 查看服务启动时间

文章目录 linux 查看服务启动时间参数解析 linux 查看服务启动时间 [root104 ~]# ps -o lstart -p ps -ef |grep -v grep |grep "zookeeper"|awk {print$2}STARTED Fri Dec 15 16:54:10 2023参数解析 linux 命令中 ps -ef 详解 ps -ef表示查看全格式的进程。 ps …...

[RK-Linux] 移植Linux-5.10到RK3399(六)| 检查GMAC(RTL8211F)配置使能千兆以太网

ROC-RK3399-PC Pro 使用 RTL8211F PHY 芯片作为以太网收发器。 RTL8211F是一种高性能的千兆以太网物理层收发器(PHY),广泛用于台式机、笔记本电脑、网络交换机等设备中。主要特点: 采用低功耗28nm CMOS技术,功耗低。支持千兆速率(10/100/1000Mbps)。支持全双工和半双工…...

博途WinCC专业版C/S架构入门指南

WinCC Professional V16 支持客户机/服务器架构&#xff0c;但目前只支持单个服务器或单对冗余服务器/多个客户机的模式&#xff0c;还不能支持像WinCC V7.5 SP1中的多个服务器/多个客户机的分布式架构。 博途工控人平时在哪里技术交流博途工控人社群 博途工控人平时在哪里技…...

大数据生态圈kafka在物联网中的应用测试

背景 由物联网项目中使用到了Tbox应用管理车辆&#xff0c;在上报数据的过程中&#xff0c;需要将终端产生的数据通过kafka的produce topic customer对数据进行处理后&#xff0c;放置到mysql中。完成数据二进制到json转换工作。 Kafka的使用 查看kafka的topic ./kafka-topi…...

ChatGPT使用:一个发包机器人的提示词

发包机器人&#xff1a; 设想&#xff1a;目前项目组有n条打包线会输出多个包&#xff0c;用户想获取最新的包是比较困难的&#xff0c;难点在于 1. 分支多&#xff1a;trunk&#xff0c;release&#xff0c;outer等&#xff0c;至少有3个分支&#xff1b; 2. 多平台&#x…...

Axure元件库的使用

1.基本元件库 1.1Axure的画布范围 Axure是一个绘制项目原型图的软件&#xff0c;它里面的基本原件有&#xff1a; 1.1元件的呈现范围 首先我们要了解基本元件的作用范围在哪里&#xff1f; 浏览效果&#xff1a; 可以看出当我们的基本元件放在画布区域内是可以完全呈现出来…...

Unity中Shader URP最简Shader框架(整理总结篇)

文章目录 前言一、精简 ShaderGraph 所有冗余代码后的最简 URP Shader二、我们来对比一下 URP Shader 与 BuildInRP Shader 的对应关系 与 区别1、"RenderPipeline""UniversalPipeline"2、面片剔除、深度测试、深度写入、颜色混合 和 BRP 下一致3、必须引入…...

AT32F435飞控之DIATONE MAMBA MK5 F435 Anti-Interference

AT32F435飞控之DIATONE MAMBA MK5 F435 Anti-Interference 1. 源由2. 规格3. 分析3.1 喜欢3.2 不便3.3 建议 4. 总结5. 参考资料 1. 源由 AT32 F435飞控在xFlight开源飞控之AT32F435计划一文中已经大体阐述了一些移植历史。 之前整体上看&#xff0c;就是航模飞控新MCU的移植…...

ntp时间同步配置中 server、pool和peer的区别

在 NTP&#xff08;Network Time Protocol&#xff09;的配置中&#xff0c;server、pool 和 peer 是用于指定时间同步关系的关键字&#xff0c;它们在角色和行为上有一些区别。 server&#xff1a; server 关键字用于指定一个或多个 NTP 服务器&#xff0c;这些服务器将提供时…...

JMeter安装RabbitMQ测试插件

整体流程如下&#xff1a;先下载AMQP插件源码&#xff0c;可以通过antivy在本地编译成jar包&#xff0c;再将jar包导入JMeter目录下&#xff0c;重启JMeter生效。 Apache Ant 是一个基于 Java 的构建工具。Ant 可用于自动化构建和部署 Java 应用程序&#xff0c;使开发人员更轻…...

基于ssm日用品网站设计论文

摘 要 现代经济快节奏发展以及不断完善升级的信息化技术&#xff0c;让传统数据信息的管理升级为软件存储&#xff0c;归纳&#xff0c;集中处理数据信息的管理方式。本日用品网站就是在这样的大环境下诞生&#xff0c;其可以帮助管理者在短时间内处理完毕庞大的数据信息&…...

coco数据集格式的RandomCrop

transforms.py文件的改进 添加 RandomCrop 函数 class RandomCrop(object):"""随机裁剪图像以及bboxes"""def __init__(self, output_size):self.output_size output_sizedef __call__(self, image, target):height, width image.shape[-2:]…...

机器学习-KL散度的直观理解+代码

KL散度 直观理解&#xff1a;KL散度是一种衡量两个分布之间匹配程度的方法。通常在概率和统计中&#xff0c;我们会用更简单的近似分布来代替观察到的数据或复杂的分布&#xff0c;KL散度帮我们衡量在选择近似值时损失了多少信息。 在信息论或概率论中&#xff0c;KL散度&#…...

【教程】制作 iOS 推送证书

​ 目录 证书类型 MAC Key Store 消息推送控制台 制作证书 创建苹果 App ID 使用appuploder制作 .p12文件 创建证书 如需向 iOS 设备推送数据&#xff0c;您首先需要在消息推送控制台上配置 iOS 推送证书。iOS 推送证书用于推送通知&#xff0c;本文将介绍消息推送服务支…...

ToolLLM model 以及LangChain AutoGPT Xagent在调用外部工具Tools的表现对比浅析

文章主要谈及主流ToolLLM 以及高口碑Agent 在调用Tools上的一些对比&#xff0c;框架先上&#xff0c;内容会不断丰富与更新。 第一部分&#xff0c;ToolLLM model 先来说主打Function Call 的大模型们 OpenAI GPT 宇宙第一LLM&#xff0c;它的functionCall都知道&#xff0…...

【MySQL学习之基础篇】约束

文章目录 1. 概述2. 基础约束3. 外键约束3.1. 介绍3.2. 外键的添加3.3. 外键删除和更新行为 1. 概述 概念&#xff1a; 约束是作用于表中字段上的规则&#xff0c;用于限制存储在表中的数据。     目的&#xff1a; 保证数据库中数据的正确、有效性和完整性。 分类&#x…...

【DataSophon】大数据管理平台DataSophon-1.2.1基本使用

&#x1f984; 个人主页——&#x1f390;开着拖拉机回家_Linux,大数据运维-CSDN博客 &#x1f390;✨&#x1f341; &#x1fa81;&#x1f341;&#x1fa81;&#x1f341;&#x1fa81;&#x1f341;&#x1fa81;&#x1f341; &#x1fa81;&#x1f341;&#x1fa81;&am…...

基于redisson实现发布订阅(多服务间用避坑)

前言 今天要分享的是基于Redisson实现信息发布与订阅&#xff08;以前分享过直接基于redis的实现&#xff09;&#xff0c;如果你是在多服务间基于redisson做信息传递&#xff0c;并且有服务压根就收不到信息&#xff0c;那你一定要看完。 今天其实重点是避坑&#xff0…...

Java 源码、反码、补码 位运算

文章目录 1. 源码、反码、补码1.1 原码1.2 反码1.3 补码1.4 byte的最大值1.5 byte的最小值 2. 位运算2.1 & 与2.2 | 或2.3 ~ 非2.4 ^ 异或2.5 << 左移 &#xff08;没有无符号左移&#xff09;2.6 >> 右移 &#xff08;有符号右移&#xff09;2.7 >>>…...

2025 后端自学UNIAPP【项目实战:旅游项目】6、我的收藏页面

代码框架视图 1、先添加一个获取收藏景点的列表请求 【在文件my_api.js文件中添加】 // 引入公共的请求封装 import http from ./my_http.js// 登录接口&#xff08;适配服务端返回 Token&#xff09; export const login async (code, avatar) > {const res await http…...

Neo4j 集群管理:原理、技术与最佳实践深度解析

Neo4j 的集群技术是其企业级高可用性、可扩展性和容错能力的核心。通过深入分析官方文档,本文将系统阐述其集群管理的核心原理、关键技术、实用技巧和行业最佳实践。 Neo4j 的 Causal Clustering 架构提供了一个强大而灵活的基石,用于构建高可用、可扩展且一致的图数据库服务…...

大模型多显卡多服务器并行计算方法与实践指南

一、分布式训练概述 大规模语言模型的训练通常需要分布式计算技术,以解决单机资源不足的问题。分布式训练主要分为两种模式: 数据并行:将数据分片到不同设备,每个设备拥有完整的模型副本 模型并行:将模型分割到不同设备,每个设备处理部分模型计算 现代大模型训练通常结合…...

html-<abbr> 缩写或首字母缩略词

定义与作用 <abbr> 标签用于表示缩写或首字母缩略词&#xff0c;它可以帮助用户更好地理解缩写的含义&#xff0c;尤其是对于那些不熟悉该缩写的用户。 title 属性的内容提供了缩写的详细说明。当用户将鼠标悬停在缩写上时&#xff0c;会显示一个提示框。 示例&#x…...

USB Over IP专用硬件的5个特点

USB over IP技术通过将USB协议数据封装在标准TCP/IP网络数据包中&#xff0c;从根本上改变了USB连接。这允许客户端通过局域网或广域网远程访问和控制物理连接到服务器的USB设备&#xff08;如专用硬件设备&#xff09;&#xff0c;从而消除了直接物理连接的需要。USB over IP的…...

Android第十三次面试总结(四大 组件基础)

Activity生命周期和四大启动模式详解 一、Activity 生命周期 Activity 的生命周期由一系列回调方法组成&#xff0c;用于管理其创建、可见性、焦点和销毁过程。以下是核心方法及其调用时机&#xff1a; ​onCreate()​​ ​调用时机​&#xff1a;Activity 首次创建时调用。​…...

力扣热题100 k个一组反转链表题解

题目: 代码: func reverseKGroup(head *ListNode, k int) *ListNode {cur : headfor i : 0; i < k; i {if cur nil {return head}cur cur.Next}newHead : reverse(head, cur)head.Next reverseKGroup(cur, k)return newHead }func reverse(start, end *ListNode) *ListN…...

解读《网络安全法》最新修订,把握网络安全新趋势

《网络安全法》自2017年施行以来&#xff0c;在维护网络空间安全方面发挥了重要作用。但随着网络环境的日益复杂&#xff0c;网络攻击、数据泄露等事件频发&#xff0c;现行法律已难以完全适应新的风险挑战。 2025年3月28日&#xff0c;国家网信办会同相关部门起草了《网络安全…...

从 GreenPlum 到镜舟数据库:杭银消费金融湖仓一体转型实践

作者&#xff1a;吴岐诗&#xff0c;杭银消费金融大数据应用开发工程师 本文整理自杭银消费金融大数据应用开发工程师在StarRocks Summit Asia 2024的分享 引言&#xff1a;融合数据湖与数仓的创新之路 在数字金融时代&#xff0c;数据已成为金融机构的核心竞争力。杭银消费金…...

PHP 8.5 即将发布:管道操作符、强力调试

前不久&#xff0c;PHP宣布了即将在 2025 年 11 月 20 日 正式发布的 PHP 8.5&#xff01;作为 PHP 语言的又一次重要迭代&#xff0c;PHP 8.5 承诺带来一系列旨在提升代码可读性、健壮性以及开发者效率的改进。而更令人兴奋的是&#xff0c;借助强大的本地开发环境 ServBay&am…...