ClickHouse10-ClickHouse中Kafka表引擎
Kafka表引擎也是一种常见的表引擎,在很多大数据量的场景下,会从源通过Kafka将数据输送到ClickHouse,Kafka作为输送的方式,ClickHouse作为存储引擎与查询引擎,大数据量的数据可以得到快速的、高压缩的存储。
Kafka大家肯定不陌生:
- 它可以用于发布和订阅数据流,是常见的队列使用方式
- 它可以组织容错存储,是常见的容错存储的使用方式
- 它可以在流可用时对其进行处理,是常见的大数据处理的使用方式
全文概览:
- 基本语法
- 从 Kafka 写入到 ClickHouse
- 从 ClickHouse 写入到 Kafka
- 测试1:queue->ck->queue
- 测试2:ck->queue
基本语法
分为定义表结构和定义Kafka的接入参数,Kafka的接入参数都是常见的字段
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(name1 [type1] [ALIAS expr1],name2 [type2] [ALIAS expr2],...
) ENGINE = Kafka()
SETTINGSkafka_broker_list = 'host:port',kafka_topic_list = 'topic1,topic2,...',kafka_group_name = 'group_name',kafka_format = 'data_format'[,][kafka_schema = '',][kafka_num_consumers = N,][kafka_max_block_size = 0,][kafka_skip_broken_messages = N,][kafka_commit_every_batch = 0,][kafka_client_id = '',][kafka_poll_timeout_ms = 0,][kafka_poll_max_batch_size = 0,][kafka_flush_interval_ms = 0,][kafka_thread_per_consumer = 0,][kafka_handle_error_mode = 'default',][kafka_commit_on_select = false,][kafka_max_rows_per_message = 1];
示例:
CREATE TABLE IF NOT EXISTS test_ck_sync1
(`sys_time` Datetime COMMENT '',`num` UInt32 COMMENT ''
)
ENGINE = Kafka
SETTINGS kafka_broker_list = '127.0.0.1:9092', kafka_topic_list = 'test_ck_sync1', kafka_group_name = 'ck_test_ck_sync1', kafka_format = 'CSV', kafka_max_block_size = 200000, kafka_skip_broken_messages = 1000, kafka_row_delimiter = '\n', format_csv_delimiter = '|'
从 Kafka 写入到 ClickHouse
创建topic:
bin/kafka-topics.sh --create --bootstrap-server 127.0.0.1:9092 --replication-factor 1 --partitions 1 --topic test_ck_sync1
创建同步表:
CREATE TABLE IF NOT EXISTS test_ck_sync1
(`sys_time` Datetime COMMENT '',`num` UInt32 COMMENT ''
)
ENGINE = Kafka
SETTINGS kafka_broker_list = '127.0.0.1:9092', kafka_topic_list = 'test_ck_sync1', kafka_group_name = 'ck_test_ck_sync1', kafka_format = 'CSV', kafka_max_block_size = 200000, kafka_skip_broken_messages = 1000, kafka_row_delimiter = '\n', format_csv_delimiter = '|'CREATE TABLE IF NOT EXISTS test_ck_sync1_res
(`sys_time` Datetime COMMENT '',`num` UInt32 COMMENT ''
)
ENGINE = MergeTree
PARTITION BY toYYYYMMDD(sys_time)
ORDER BY tuple()
创建物化视图,进行数据样式的转换:
CREATE MATERIALIZED VIEW test_ck_sync1_mv TO test_ck_sync1_res AS
SELECTsys_time,num
FROM test_ck_sync1
通过console写入数据:
[$ kafka_2.13-3.6.1]# bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic test_ck_sync1
>2024-01-01 00:00:01|89
验证数据:
$ :) select * from test_ck_sync1_res;SELECT *
FROM test_ck_sync1_resQuery id: a666f893-5be9-4022-9327-3a1507aa5485┌────────────sys_time─┬─num─┐
│ 2024-01-01 00:00:01 │ 89 │
└─────────────────────┴─────┘
┌────────────sys_time─┬─num─┐
│ 2024-01-01 00:00:00 │ 88 │
└─────────────────────┴─────┘2 rows in set. Elapsed: 0.049 sec.
从 ClickHouse 写入到 Kafka
kafka_writers_reader --(view)--> kafka_writers_queue --->
创建一个队列:
bin/kafka-topics.sh --topic kafka_writers --create -bootstrap-server 127.0.0.1:9092 --partitions 1 --replication-factor 1
创建同步表:
CREATE TABLE kafka_writers_reader ( `id` Int, `platForm` String, `appname` String, `time` DateTime )
ENGINE = Kafka SETTINGS kafka_broker_list = '127.0.0.1:9092', kafka_topic_list = 'kafka_writers_reader', kafka_group_name = 'kafka_writers_reader_group', kafka_format = 'CSV';CREATE TABLE kafka_writers_queue ( id Int, platForm String, appname String, time DateTime )
ENGINE = Kafka SETTINGS kafka_broker_list = '127.0.0.1:9092', kafka_topic_list = 'kafka_writers', kafka_group_name = 'kafka_writers_group', kafka_format = 'CSV', kafka_max_block_size = 1048576;
测试1:queue->ck->queue
通过写入队列kafka_writers_reader,借助ClickHouse写入队列kafka_writers
bin/kafka-topics.sh --topic kafka_writers_reader --create -bootstrap-server 127.0.0.1:9092 --partitions 1 --replication-factor 1bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic kafka_writers_readerbin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic kafka_writers
测试2:ck->queue
通过写入表kafka_writers_reader,写入队列kafka_writers
$ :) INSERT INTO kafka_writers_reader (id, platForm, appname, time)
VALUES (8,'Data','Test','2020-12-23 14:45:31'),
(9,'Plan','Test1','2020-12-23 14:47:32'),
(10,'Plan','Test2','2020-12-23 14:52:15'),
(11,'Data','Test3','2020-12-23 14:54:39');INSERT INTO kafka_writers_reader (id, platForm, appname, time) FORMAT ValuesQuery id: 223a63ab-97fa-488d-8ea7-c2e194155d26Ok.4 rows in set. Elapsed: 1.054 sec.
[$ kafka_2.13-3.6.1]# bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic kafka_writers
8,"Data","Test","1970-01-01 08:00:00"9,"Plan","Test1","1970-01-01 08:00:00"10,"Plan","Test2","1970-01-01 08:00:00"11,"Data","Test3","1970-01-01 08:00:00"
如果喜欢我的文章的话,可以去GitHub上给一个免费的关注吗?
相关文章:

ClickHouse10-ClickHouse中Kafka表引擎
Kafka表引擎也是一种常见的表引擎,在很多大数据量的场景下,会从源通过Kafka将数据输送到ClickHouse,Kafka作为输送的方式,ClickHouse作为存储引擎与查询引擎,大数据量的数据可以得到快速的、高压缩的存储。 Kafka大家…...
Encoding类
Encoding System.Text.Encoding 是 C# 中用于处理字符编码和字符串与字节之间转换的类。它提供了各种静态方法和属性,**用于在不同字符编码之间进行转换,**以及将字符串转换为字节数组或反之。 在处理多语言文本、文件、网络通信以及其他字符数据的场景…...

标定系列——预备知识-OpenCV中实现Rodrigues变换的函数(二)
标定系列——预备知识-OpenCV中实现Rodrigues变换的函数(二) 说明记录 说明 简单介绍罗德里格斯变换以及OpenCV中的实现函数 记录...

2014年认证杯SPSSPRO杯数学建模C题(第一阶段)土地储备方案的风险评估全过程文档及程序
2014年认证杯SPSSPRO杯数学建模 C题 土地储备方案的风险评估 原题再现: 土地储备,是指市、县人民政府国土资源管理部门为实现调控土地市场、促进土地资源合理利用目标,依法取得土地,进行前期开发、储存以备供应土地的行为。土地…...

我的编程之路:从非计算机专业到Java开发工程师的成长之路 | 学习路线 | Java | 零基础 | 学习资源 | 自学
小伙伴们好,我是「 行走的程序喵」,感谢您阅读本文,欢迎三连~ 😻 【Java基础】专栏,Java基础知识全面详解:👉点击直达 🐱 【Mybatis框架】专栏,入门到基于XML的配置、以…...

Django Cookie和Session
Django Cookie和Session 【一】介绍 【1】起因 HTTP协议四大特性 基于请求响应模式:客户端发送请求,服务端返回响应基于TCP/IP之上:作用于应用层之上的协议无状态:HTTP协议本身不保存客户端信息短链接:1.0默认使用短…...

【算法刷题 | 二叉树 04】3.27(翻转二叉树、对称二叉树、完全二叉树的节点个数、平衡二叉树、完全二叉树的所有路径)
文章目录 6.翻转二叉树6.1问题6.2解法一:递归6.2.1递归思路(1)确定递归函数的参数和返回值(2)确定终止条件(3)确定单层递归的逻辑 6.2.2全部代码 6.3解法二:层序遍历 7.对称二叉树7.…...

【uniapp】uniapp实现免密登录
文章目录 一、概要二、整体架构流程三、技术名词解释四 、技术细节1.存取token有效期?2.使用setStorageSync而不使用setStorage?3.使用onLaunch而不使用全局路由? 一、概要 打开一个网页或小程序的时候,我们有时候会自动进入主页…...
2024-03-27 问AI: 介绍一下深度学习中的 Darknet
文心一言 在深度学习中,Darknet是一个开源的神经网络框架,由Joseph Redmon开发,主要用于实现深度学习算法。它以其高效的实现和速度受到广泛关注,并在许多计算机视觉竞赛中取得了优异的结果。 Darknet的特点包括: 轻…...

POJ3037 + HDU-6714
两道最短路好题 POJ3037 手玩一下 发现每一点的速度可以直接搞出来,就是pow(2,h[1][1]-h[i][j])*V 那么从这个点出发到达别的点的耗费的时间都是上面这个数的倒数,然后直接跑最短路就好了 #include<iostream> #include<vector> #include<…...
Ubuntu搭建环境Cmake-Libtorch-Torchvision-PCL-VTK-OpenCV
Ubuntu搭建环境Cmake-Libtorch-Torchvision-PCL-VTK-OpenCV 安装Cmake安装libtorch安装torchvision安装PCL安装VTK安装OpenCV设置环境变量 仅供本人记录查阅使用 安装Cmake Cmake下载地址 解压 进入目录会看到只有 bin doc man share三个文件夹,没有 bootstrap文…...

分享多种mfc100u.dll丢失的解决方法(一键修复DLL丢失的方法)
在使用电脑过程中,我们经常会遇到一些陌生的DLL文件,例如mfc100u.dll。这些DLL文件是动态链接库(Dynamic Link Libraries)的缩写,它们包含了可以被多个程序共享的代码和数据。今天,我们将深入探讨mfc100u.d…...

Redis是单线程还是多线程?(面试题)
1、Redis5及之前是单线程版本 2、Redis6开始引入多线程版本(实际上是 单线程多线程 版本) Redis6及之前版本(单线程) Redis5及之前的版本使用的是 单线程,也就是说只有一个 worker队列,所有的读写操作都要…...

动态菜单设计
需求: 登录不同用户 显示不同的菜单 思路:根据用户id 左关联表 查询出对应的菜单选项 查询SQL select distinct-- 菜单表 去除重复记录sys_menu.id,sys_menu.parentId, sys_menu.name from -- 权限表sys_menu-- 角色与权限表 菜单表id 角色菜…...

Haproxy负载均衡介绍即部署
haproxy的原理: 提供高可用、负载均衡以及基于TCP(四层)和HTTP(七层)应用的代理,支持虚拟主机,开源可靠的一款软件。 适用于哪些负载特别大的web站点,这些站点通常又需要回话保持和七…...

基于大语言模型的云故障根因分析|顶会EuroSys24论文
*马明华 微软主管研究员 2021年CCF国际AIOps挑战赛程序委员会主席(第四届) 2021年博士毕业于清华大学,2020年在佐治亚理工学院做访问学者。主要研究方向是智能运维(AIOps)、软件可靠性。近年来在ICSE、FSE、ATC、EuroS…...

Windows直接运行python程序
Windows直接运行python程序 一、新建bat脚本二、新建vbs脚本 一、新建bat脚本 新建bat批处理脚本,写入以下内容 echo off call conda activate pytorch python app.pyecho off:在此语句后所有运行的命令都不显示命令行本身,但是本身的指令是…...

经典应用丨光伏行业扫码追溯新标杆,海康机器人AI智能读码器!
去年,光伏发电行业持续高速发展,我国仅在前九个月累计装机521.08GW,同比增长达到45.3%,已成为第二大电源类型超过水电。根据《2023中国与全球光伏发展白皮书》预测,到2030年,中国能够实现国家规划的风电和光…...
逆流而上的选择-积极生活,逆流而上
首先请大家看一个故事 李明坐在公司的开放式办公区,耳边是键盘敲击声的交响乐,眼前是一行行跳跃的代码。他的眼神有些恍惚,显示器的蓝光在他眼镜上反射出时代的光芒,这光芒既耀眼又刺眼。他即将35岁,在这个年纪&#x…...
SpringMVC基础Controller
文章目录 Controller 的编写和配置1. Controller 注解类型2. RequestMapping 注解类型3. 编写请求方法4. 请求参数和路径变量 Controller 的编写和配置 Controller 注解和 RequestMapping 注解是 Spring MVC 最重要的两个注解。 使用基于注解的控制器的优点如下: …...

8k长序列建模,蛋白质语言模型Prot42仅利用目标蛋白序列即可生成高亲和力结合剂
蛋白质结合剂(如抗体、抑制肽)在疾病诊断、成像分析及靶向药物递送等关键场景中发挥着不可替代的作用。传统上,高特异性蛋白质结合剂的开发高度依赖噬菌体展示、定向进化等实验技术,但这类方法普遍面临资源消耗巨大、研发周期冗长…...

关于nvm与node.js
1 安装nvm 安装过程中手动修改 nvm的安装路径, 以及修改 通过nvm安装node后正在使用的node的存放目录【这句话可能难以理解,但接着往下看你就了然了】 2 修改nvm中settings.txt文件配置 nvm安装成功后,通常在该文件中会出现以下配置&…...

让AI看见世界:MCP协议与服务器的工作原理
让AI看见世界:MCP协议与服务器的工作原理 MCP(Model Context Protocol)是一种创新的通信协议,旨在让大型语言模型能够安全、高效地与外部资源进行交互。在AI技术快速发展的今天,MCP正成为连接AI与现实世界的重要桥梁。…...

JUC笔记(上)-复习 涉及死锁 volatile synchronized CAS 原子操作
一、上下文切换 即使单核CPU也可以进行多线程执行代码,CPU会给每个线程分配CPU时间片来实现这个机制。时间片非常短,所以CPU会不断地切换线程执行,从而让我们感觉多个线程是同时执行的。时间片一般是十几毫秒(ms)。通过时间片分配算法执行。…...

html css js网页制作成品——HTML+CSS榴莲商城网页设计(4页)附源码
目录 一、👨🎓网站题目 二、✍️网站描述 三、📚网站介绍 四、🌐网站效果 五、🪓 代码实现 🧱HTML 六、🥇 如何让学习不再盲目 七、🎁更多干货 一、👨…...

Web后端基础(基础知识)
BS架构:Browser/Server,浏览器/服务器架构模式。客户端只需要浏览器,应用程序的逻辑和数据都存储在服务端。 优点:维护方便缺点:体验一般 CS架构:Client/Server,客户端/服务器架构模式。需要单独…...

Ubuntu Cursor升级成v1.0
0. 当前版本低 使用当前 Cursor v0.50时 GitHub Copilot Chat 打不开,快捷键也不好用,当看到 Cursor 升级后,还是蛮高兴的 1. 下载 Cursor 下载地址:https://www.cursor.com/cn/downloads 点击下载 Linux (x64) ,…...
在树莓派上添加音频输入设备的几种方法
在树莓派上添加音频输入设备可以通过以下步骤完成,具体方法取决于设备类型(如USB麦克风、3.5mm接口麦克风或HDMI音频输入)。以下是详细指南: 1. 连接音频输入设备 USB麦克风/声卡:直接插入树莓派的USB接口。3.5mm麦克…...

Springboot 高校报修与互助平台小程序
一、前言 随着我国经济迅速发展,人们对手机的需求越来越大,各种手机软件也都在被广泛应用,但是对于手机进行数据信息管理,对于手机的各种软件也是备受用户的喜爱,高校报修与互助平台小程序被用户普遍使用,为…...

基于Python的气象数据分析及可视化研究
目录 一.🦁前言二.🦁开源代码与组件使用情况说明三.🦁核心功能1. ✅算法设计2. ✅PyEcharts库3. ✅Flask框架4. ✅爬虫5. ✅部署项目 四.🦁演示效果1. 管理员模块1.1 用户管理 2. 用户模块2.1 登录系统2.2 查看实时数据2.3 查看天…...