【Flink-Sql-Kafka-To-ClickHouse】使用 FlinkSql 将 Kafka 数据写入 ClickHouse

【Flink-Sql-Kafka-To-ClickHouse】使用 FlinkSql 将 Kafka 数据写入 ClickHouse

  • 1)需求分析
  • 2)功能实现
  • 3)准备工作
    • 3.1.Kafka
    • 3.2.ClickHouse
  • 4)Flink-Sql
  • 5)验证

1)需求分析

1、数据源为 Kafka,定义 Kafka-Topic 为动态临时视图表。

2、写入到 ClickHouse,自定义 Sink 表。

3、source 和 sink 都使用 Flink 集成的 Connector

2)功能实现

导入ClickHouse连接器

<dependency><groupId>com.aliyun</groupId><artifactId>flink-connector-clickhouse</artifactId><version>1.14.0</version>
</dependency>

如果在服务器上执行,需要将 jar 放到 Flink 的 lib 目录下。

3)准备工作

3.1.Kafka

1、创建好Topic

2、准备测试数据

{"id": 1,"eventId": "TEST123","eventStDt": "2022-11-3023:37:49","bak6": "测试","bak7": "https://test?user","businessId": "17279811111111111111111111111111","phone": "12345678910","bak1": "1234","bak2": "2022-12-0100:00:00","bak13": "17279811111111111111111111111111","bak14": "APP","bak11": "TEST"
}

3.2.ClickHouse

1、创建表(此处我们使用生产环境中较为常用的 cluster 集群模式建表)

注意集群模式要创建两次表,一次为 local 本地表,一次为 cluster 集群表。

  • local
CREATE TABLE test.kafka2ck_test_local on cluster test_cluster 
(`id` UInt32,`eventId` LowCardinality(Nullable(String)),`eventStDt` LowCardinality(Nullable(String)),`bak6` LowCardinality(Nullable(String)),`bak7` LowCardinality(Nullable(String)),`businessId` LowCardinality(Nullable(String)),`phone` LowCardinality(Nullable(String)),`bak1` LowCardinality(Nullable(String)),`bak2` LowCardinality(Nullable(String)),`bak13` LowCardinality(Nullable(String)),`bak14` LowCardinality(Nullable(String)),`bak11` LowCardinality(Nullable(String))
)
ENGINE = ReplicatedMergeTree
PARTITION BY id
PRIMARY KEY id
ORDER BY id
SETTINGS index_granularity = 8192;
  • cluster
CREATE TABLE test.kafka2ck_test on cluster test_cluster 
(`id` UInt32,`eventId` LowCardinality(Nullable(String)),`eventStDt` LowCardinality(Nullable(String)),`bak6` LowCardinality(Nullable(String)),`bak7` LowCardinality(Nullable(String)),`businessId` LowCardinality(Nullable(String)),`phone` LowCardinality(Nullable(String)),`bak1` LowCardinality(Nullable(String)),`bak2` LowCardinality(Nullable(String)),`bak13` LowCardinality(Nullable(String)),`bak14` LowCardinality(Nullable(String)),`bak11` LowCardinality(Nullable(String))
)
ENGINE = Distributed('test_cluster', 'test', 'kafka2ck_test_local', rand());

4)Flink-Sql

  • source
CREATE TABLE source_kafka_test (id INT,eventId STRING,eventStDt STRING,bak6 STRING,bak7 STRING,businessId STRING,phone STRING,bak1 STRING,bak2 STRING,bak13 STRING,bak14 STRING,bak11 STRING) WITH ('connector' = 'kafka','topic' = 'test','format'='json','properties.bootstrap.servers' = '${kafka-bootstrap-server}','properties.group.id' = 'test01','scan.startup.mode' = 'earliest-offset','properties.security.protocol' = 'SASL_PLAINTEXT','properties.sasl.kerberos.service.name' = 'kafka'
);
  • sink
CREATE TABLE sink_ck_test (id INT,eventId STRING,eventStDt STRING,bak6 STRING,bak7 STRING,businessId STRING,phone STRING,bak1 STRING,bak2 STRING,bak13 STRING,bak14 STRING,bak11 STRING,PRIMARY KEY (id) NOT ENFORCED) WITH ('connector' = 'clickhouse','url' = 'jdbc:clickhouse://123.1.1.1:9090','database-name'='test','table-name' = 'kafka2ck_test_local','username' = 'test','password' = '123456','sink.batch-size' = '100','sink.flush-interval' = '1000','sink.max-retries' = '3'
);
  • insert
insert into sink_ck_test select * from source_kafka_test;

5)验证

在 Kafka 中写入对应 ClickHouse 格式的 Json 测试数据,观察 ClickHouse 中是否有数据写入。

{"id": 1,"eventId": "TEST123","eventStDt": "2022-11-3023:37:49","bak6": "测试","bak7": "https://test?user","businessId": "17279811111111111111111111111111","phone": "12345678910","bak1": "1234","bak2": "2022-12-0100:00:00","bak13": "17279811111111111111111111111111","bak14": "APP","bak11": "TEST"
}

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

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

相关文章

每日一博 - Cache Miss Attack

文章目录 概述解决思路缓存空值键并设置短期 TTL&#xff08;生存时间&#xff09;使用布隆过滤器 伪代码1. 缓存空值键并设置短期 TTLa. 缓存空值键b. 设置短期 TTL 2. 使用布隆过滤器a. 集成布隆过滤器b. 查询布隆过滤器 进一步优化系统性能的建议 概述 在缓存管理中&#x…

基于VGG-16+Android+Python的智能车辆驾驶行为分析—深度学习算法应用(含全部工程源码)+数据集+模型(三)

目录 前言总体设计系统整体结构图系统流程图 运行环境模块实现1. 数据预处理2. 模型构建3. 模型训练及保存1&#xff09;模型训练2&#xff09;模型保存 4. 模型生成1&#xff09;模型导入及调用2&#xff09;相关代码&#xff08;1&#xff09;布局文件&#xff08;2&#xff…

bug-ku--计算器

F12 maxlength"1" 限制的是你能输入几位数 改成3就行 来那个数相相加就能输入了 flag{464f5f406e7e182014500fc49f7aedfc}

Mysql和Oracle的区别

MySQL 和 Oracle 是两种常用的关系型数据库管理系统&#xff08;RDBMS&#xff09;&#xff0c;它们有许多相似之处&#xff0c;也有一些显著的不同点。 相同点&#xff1a; 关系型数据库&#xff1a; MySQL 和 Oracle 都是关系型数据库&#xff0c;采用表&#xff08;table&a…

Vue3+Ts项目——登录页面跳转到首页

前言&#xff1a;根据我上篇所实现的左边菜单栏之后&#xff0c;需要登录成功跳转home页面。主要分为三步。 第一步&#xff1a;创建三个ts文件结合pinia缓存登录信息和token src\api\userTypes.ts 就是个接口类方便页面和另一个ts文件数据传递&#xff0c;其实也可以不用 …

【异步绘制】UIView刷新原理 与 异步绘制

快捷目录 壹、 iOS界面刷新机制贰、浅谈UIView的刷新与绘制概述一.UIView 与 CALayer1. UIView 与 CALayer的关系2. CALayer的一些常用属性contents属性contentGravity属性contentsScale属性maskToBounds属性contentsRect属性 二.View的布局与显示1.图像显示原理2.布局layoutSu…

【UE】在蓝图中修改材质实例的参数的两种方式

目录 方式一、通过“在材质上设置标量/向量参数值”节点实现 方式二、通过“设置标量/向量参数值”节点实现 方式一、通过“在材质上设置标量/向量参数值”节点实现 1. 在材质中设置了两个参数 2. 创建材质实例 3. 创建一个蓝图&#xff0c;对静态网格体赋予材质实例 在事件…

【C++初阶】类与对象(上)

类与对象&#xff08;上&#xff09; 1.面向过程和面向对象初步认识2.类的引入3.类的定义4.类的访问限定符及封装4.1 访问限定符4.2 封装 5.类的作用域6.类的实例化7.类对象模型7.1 如何计算类对象的大小7.2 结构体内存对齐规则 8.this指针8.1 this指针的引出8.2 this指针的特性…

C语言——高精度减法

相较于高精度加法&#xff0c;高精度减法会有一些不同&#xff0c;如果是num1 - num2的话&#xff0c;如果num1 > num2时&#xff0c;也就是结果为正时&#xff0c;只需要对每一位进行按位相减&#xff0c;如果不够就向下一位借位&#xff1b;但是如果是num1 < num2时&am…

【算法与数据结构】53、LeetCode最大子数组和

文章目录 一、题目二、解法三、完整代码 所有的LeetCode题解索引&#xff0c;可以看这篇文章——【算法和数据结构】LeetCode题解。 一、题目 二、解法 思路分析&#xff1a;程序一共两个变量&#xff0c;一个result一个count。result用来记录最终的结果&#xff0c;count记录当…

LAMP平台——构建PHP运行环境

在构建LAMP平台时&#xff0c;各组件的安装顺序依次为Linux、Apache、MySQL、PHP。其中Apache和 MySQL的安装并没有严格的顺序&#xff1b;而PHP环境的安装一般放到最后&#xff0c;负责沟通Web服务器和数据库 系统以协同工作。 PHP 即 Hypertext Preprocessor&#xff08;超级…

数据结构 之map/set练习

文章目录 1. 只出现一次的数字算法原理&#xff1a;代码&#xff1a; 2. 随机链表的复制算法原理&#xff1a;代码&#xff1a; 3. 宝石与石头算法原理&#xff1a;代码&#xff1a; 4. 坏键盘打字算法原理&#xff1a;代码&#xff1a; 5. 前K个高频单词算法原理&#xff1a;代…

图片的批量建码怎么做?一图一码的制作方法

在使用图片展示内容时&#xff0c;经常会有同一类型的图片信息是有区别的&#xff0c;如果需要将每张图片批量生成二维码图片&#xff0c;那么出了一张一张去制作之外&#xff0c;有没有能够一键批量建码的功能可以解决这个问题呢&#xff1f;下面来给大家分享一下图片批量建码…

SpringBoot 源码解析1:环境搭建

SpringBoot 源码解析1&#xff1a;环境搭建 1.项目结构2.pom.xml3.MyApplication 1.项目结构 最简单的spring-web项目 2.pom.xml <?xml version"1.0" encoding"UTF-8"?> <project xmlns"http://maven.apache.org/POM/4.0.0"xmlns…

2023 年安徽省职业院校技能大赛信息安全管理与评估项目竞赛须知

2023 年安徽省职业院校技能大赛&#xff08;高职组&#xff09; “信息安全管理与评估”项目竞赛须知 一、参赛队须知 1.参赛队名称&#xff1a;统一使用规定的学校代表队名称&#xff0c;不接受跨市、跨校组队报名。 2.参赛队组成&#xff1a;本赛项为团体赛&#xff0c;以…

记删除CK不彻底问题 及 新增表TTL机制

问题背景&#xff1a;对CK表进行删除时&#xff0c;发现无法彻底删除&#xff0c;并报错如下&#xff1a; 同时也会有下面的报错信息&#xff1a; 解决过程&#xff1a; 确认CK 节点是否健康存活&#xff0c;select * from system.clusters 可以查看&#xff1b;确认CK元数据是…

redis中存储list<map>,list<entity>结构数据如何处理

redis中存储list结构数据如何处理 用到的工具类 1、序列化操作 String multiMapStr JSON.toJSONString(multiMapList); // list转换为json字符串 Object o JSON.parse(userList); // list转换为object List<Map<String, String>> maps CastBeanUtil.castListM…

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

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

linux系统的u盘/mmc/sd卡等的支持热插拔和自动挂载行为

1.了解mdev mdev是busybox自带的一个简化版的udev。udev是从Linux 2.6 内核系列开始的设备文件系统&#xff08;DevFS&#xff09;的替代品&#xff0c;是 Linux 内核的设备管理器。总的来说&#xff0c;它取代了 devfs 和 hotplug&#xff0c;负责管理 /dev 中的设备节点。同时…

openEuler商业化进展可观:累计装机量超610万套,市场持续扩容

12月15日至16日&#xff0c;以“崛起数字时代&#xff0c;引领数智未来”为主题的操作系统大会&#xff06;openEuler Summit 2023在北京国家会议中心举办。大会旨在汇聚全球产业界创新力量&#xff0c;构筑坚实的基础软件根基&#xff0c;推动基础软件技术持续创新&#xff0c…