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 最重要的两个注解。 使用基于注解的控制器的优点如下: …...
进程地址空间(比特课总结)
一、进程地址空间 1. 环境变量 1 )⽤户级环境变量与系统级环境变量 全局属性:环境变量具有全局属性,会被⼦进程继承。例如当bash启动⼦进程时,环 境变量会⾃动传递给⼦进程。 本地变量限制:本地变量只在当前进程(ba…...

为什么需要建设工程项目管理?工程项目管理有哪些亮点功能?
在建筑行业,项目管理的重要性不言而喻。随着工程规模的扩大、技术复杂度的提升,传统的管理模式已经难以满足现代工程的需求。过去,许多企业依赖手工记录、口头沟通和分散的信息管理,导致效率低下、成本失控、风险频发。例如&#…...
c++ 面试题(1)-----深度优先搜索(DFS)实现
操作系统:ubuntu22.04 IDE:Visual Studio Code 编程语言:C11 题目描述 地上有一个 m 行 n 列的方格,从坐标 [0,0] 起始。一个机器人可以从某一格移动到上下左右四个格子,但不能进入行坐标和列坐标的数位之和大于 k 的格子。 例…...

智能在线客服平台:数字化时代企业连接用户的 AI 中枢
随着互联网技术的飞速发展,消费者期望能够随时随地与企业进行交流。在线客服平台作为连接企业与客户的重要桥梁,不仅优化了客户体验,还提升了企业的服务效率和市场竞争力。本文将探讨在线客服平台的重要性、技术进展、实际应用,并…...

从零开始打造 OpenSTLinux 6.6 Yocto 系统(基于STM32CubeMX)(九)
设备树移植 和uboot设备树修改的内容同步到kernel将设备树stm32mp157d-stm32mp157daa1-mx.dts复制到内核源码目录下 源码修改及编译 修改arch/arm/boot/dts/st/Makefile,新增设备树编译 stm32mp157f-ev1-m4-examples.dtb \stm32mp157d-stm32mp157daa1-mx.dtb修改…...
爬虫基础学习day2
# 爬虫设计领域 工商:企查查、天眼查短视频:抖音、快手、西瓜 ---> 飞瓜电商:京东、淘宝、聚美优品、亚马逊 ---> 分析店铺经营决策标题、排名航空:抓取所有航空公司价格 ---> 去哪儿自媒体:采集自媒体数据进…...
代理篇12|深入理解 Vite中的Proxy接口代理配置
在前端开发中,常常会遇到 跨域请求接口 的情况。为了解决这个问题,Vite 和 Webpack 都提供了 proxy 代理功能,用于将本地开发请求转发到后端服务器。 什么是代理(proxy)? 代理是在开发过程中,前端项目通过开发服务器,将指定的请求“转发”到真实的后端服务器,从而绕…...

Spring Cloud Gateway 中自定义验证码接口返回 404 的排查与解决
Spring Cloud Gateway 中自定义验证码接口返回 404 的排查与解决 问题背景 在一个基于 Spring Cloud Gateway WebFlux 构建的微服务项目中,新增了一个本地验证码接口 /code,使用函数式路由(RouterFunction)和 Hutool 的 Circle…...
return this;返回的是谁
一个审批系统的示例来演示责任链模式的实现。假设公司需要处理不同金额的采购申请,不同级别的经理有不同的审批权限: // 抽象处理者:审批者 abstract class Approver {protected Approver successor; // 下一个处理者// 设置下一个处理者pub…...

安宝特案例丨Vuzix AR智能眼镜集成专业软件,助力卢森堡医院药房转型,赢得辉瑞创新奖
在Vuzix M400 AR智能眼镜的助力下,卢森堡罗伯特舒曼医院(the Robert Schuman Hospitals, HRS)凭借在无菌制剂生产流程中引入增强现实技术(AR)创新项目,荣获了2024年6月7日由卢森堡医院药剂师协会࿰…...