ClickHouse Kafka 引擎教程

如果您刚开始并且第一次设置 Kafka 和 ClickHouse 需要帮助怎么办？这篇文章也许会提供下帮助。

我们将通过一个端到端示例，使用 Kafka 引擎将数据从 Kafka 主题加载到 ClickHouse 表中。我们还将展示如何重置偏移量和重新加载数据，以及如何更改表架构。最后，我们将演示如何将数据从 ClickHouse 写回 Kafka 主题。

先决条件

下面的练习假设你已经安装并运行了 Kafka 和 ClickHouse。为了方便起见，我们使用了 Kubernetes。Kafka 版本是 Confluent 5.4.0，使用带有三个 Kafka 代理的 Kafka helm chart 安装。ClickHouse版本为20.4.2，使用ClickHouse Kubernetes Operator安装在单个节点上。

这些练习应该适用于任何类型的安装，但您需要相应地更改主机名。如果 Kafka 代理较少，则可能还需要更改复制因子。

Kafka-ClickHouse 集成概述

Kafka 是一种极具可扩展性的消息总线。它的核心是由运行在不同主机上的代理管理的分布式日志。以下是应用程序模型的简短说明。

生产者将消息写入主题，主题是一组消息。使用者从主题中读取消息，该主题分布在分区上。消费者被安排在消费者组中，这允许应用程序从 Kafka 并行读取消息，而不会丢失或重复。

下图说明了上述主要部分。

ClickHouse 可以使用 Kafka 表引擎和物化视图直接从 Kafka 主题读取消息，该视图获取消息并将其推送到 ClickHouse 目标表。目标表通常使用 MergeTree 引擎或 ReplicatedMergeTree 等变体来实现。消息流如下图所示。

也可以从 ClickHouse 写回 Kafka。消息流更简单 - 只需插入到 Kafka 表中即可。下面是流程图。

在 Kafka 上创建主题

现在让我们在 Kafka 上设置一个主题，我们可以使用它来加载消息。登录到 Kafka 服务器，然后使用以下示例中的命令创建主题。在此示例中，“kafka”是服务器的 DNS 名称。如果您有其他 DNS 名称，请改用该名称。您还可以调整分区数以及复制因子。

kafka-topics \
--bootstrap-server kafka:9092 \
--topic readings \
--create --partitions 6 \
--replication-factor 2

检查主题是否已成功创建。

kafka-topics --bootstrap-server kafka:9092 --describe readings
你将看到如下所示的输出，其中显示了其分区的主题和当前状态。 
Topic: readings    PartitionCount: 6    ReplicationFactor: 2    Configs:
    Topic: readings    Partition: 0    Leader: 0    Replicas: 0,2    Isr: 0,2
    Topic: readings    Partition: 1    Leader: 2    Replicas: 2,1    Isr: 2,1
    Topic: readings    Partition: 2    Leader: 1    Replicas: 1,0    Isr: 1,0
    Topic: readings    Partition: 3    Leader: 0    Replicas: 0,1    Isr: 0,1
    Topic: readings    Partition: 4    Leader: 2    Replicas: 2,0    Isr: 2,0
    Topic: readings    Partition: 5    Leader: 1    Replicas: 1,2    Isr: 1,2

现在 Kafak 准备工作已完成。让我们转向ClickHouse。

ClickHouse Kafka 引擎设置

要将数据从 Kafka 主题读取到 ClickHouse 表，我们需要做三件事：

  • 一个目标 MergeTree 表，用于为引入的数据提供主目录

  • 一个 Kafka 引擎表，使主题看起来像一个 ClickHouse 表

  • 用于自动将数据从 Kafka 移动到目标表的具体化视图

首先，我们将定义目标 MergeTree 表。登录到 ClickHouse 执行以下 SQL

CREATE TABLE readings (
    readings_id Int32 Codec(DoubleDelta, LZ4),
    time DateTime Codec(DoubleDelta, LZ4),
    date ALIAS toDate(time),
    temperature Decimal(5,2) Codec(T64, LZ4)
) Engine = MergeTree
PARTITION BY toYYYYMM(time)
ORDER BY (readings_id, time);

接下来，我们需要使用 Kafka 引擎创建一个表来连接主题并读取数据。引擎将使用主题“readings”和消费者组名称“readings consumer_group1”从主机 kafka 的代理读取数据。输入格式为 CSV。

请注意，我们省略了“date”列。它是目标表中的别名，将从“time”列自动填充。


CREATE TABLE readings_queue (
    readings_id Int32,
    time DateTime,
    temperature Decimal(5,2)
)
ENGINE = Kafka
SETTINGS kafka_broker_list = 'kafka-headless.kafka:9092',
         kafka_topic_list = 'readings',
         kafka_group_name = 'readings_consumer_group1',
         kafka_format = 'CSV',
         kafka_max_block_size = 1048576;

前面的设置处理最简单的情况：单个代理、单个主题且没有专用配置。

最后，我们创建一个物化视图，用于在 Kafka 和合并树表之间传输数据。

CREATE MATERIALIZED VIEW readings_queue_mv TO readings AS
SELECT readings_id, time, temperature
FROM readings_queue;

这就是 Kafka 到 ClickHouse 的集成。让我们来测试一下。

加载数据

现在是时候使用 kafka-console-producer 命令加载一些输入数据了。以下示例使用 CSV 格式添加三条记录。


kafka-console-producer --broker-list kafka:9092 --topic readings <<END
1,"2020-05-16 23:55:44",14.2
2,"2020-05-16 23:55:45",20.1
3,"2020-05-16 23:55:51",12.9
END
传输到 readings  表需要几秒钟。如果我们从表中进行查询，我们会得到以下输出。 
SELECT *
FROM readings

┌─readings_id─┬────────────────time─┬─temperature─┐
│           1 │ 2020-05-16 23:55:44 │       14.20 │
│           2 │ 2020-05-16 23:55:45 │       20.10 │
│           3 │ 2020-05-16 23:55:51 │       12.90 │
└─────────────┴─────────────────────┴─────────────┘

Kafka 和 ClickHouse 现已连接。

从 Kafka 重读消息

前面的示例从 Kafka 主题中的起始位置开始，并在消息到达时读取消息。这是正常方式，但有时再次阅读消息很有用。例如，您可能希望在修复架构中的 bug 或重新加载备份后重新读取消息。幸运的是，这很容易做到。我们只是重置使用者组中的偏移量。

假设我们丢失了 readings 表中的所有消息，并希望从 Kafka 重新加载它们。首先，让我们使用 TRUNCATE 命令“丢失”消息。

TRUNCATE TABLE readings;

在重置分区上的偏移量之前，我们需要关闭消息消费。为此，请在 ClickHouse 中分离 readings_queue 表，如下所示。

DETACH TABLE readings_queue

接下来，使用以下 Kafka 命令重置用于 readings_queue 表的使用者组中的分区偏移量 (kafka 节点执行)。

kafka-consumer-groups --bootstrap-server kafka:9092 \
 --topic readings --group readings_consumer_group1 \
 --reset-offsets --to-earliest --execute

现在重新连接readings_queue表。

ATTACH TABLE readings_queue

等待几秒钟，丢失的记录将被恢复。您可以运行 SELECT 来确认它们已恢复。

添加虚拟列

使用显示原始 Kafka 消息坐标的信息标记行通常很有用。为此，Kafka 表引擎自动定义了虚拟列。下面介绍如何更改 readings 表以显示源主题、分区和偏移量。

首先，让我们通过分离 Kafka 表来禁用消息使用。消息可能会堆积在主题上，但我们不会丢失它们。

DETACH TABLE readings_queue

接下来，我们通过连续执行以下 SQL 命令来更改目标表和物化视图。请注意，我们只是删除并重新创建具体化视图，而更改目标表，从而保留现有数据。

ALTER TABLE readings
  ADD COLUMN _topic String,
  ADD COLUMN _offset UInt64,
  ADD COLUMN _partition UInt64
  
DROP TABLE readings_queue_mv

CREATE MATERIALIZED VIEW readings_queue_mv TO readings AS
  SELECT readings_id, time, temperature, _topic, _offset, _partition
  FROM readings_queue;

最后，我们通过重新附加 readings_queue 表来再次启用消息使用。


ATTACH TABLE readings_queue

您可以通过截断表并重新加载消息来确认新架构，就像我们在上一节中所做的那样。如果查询数据，它将如下所示。

SELECT
    readings_id AS id, time, temperature AS temp,
    _topic, _offset, _partition
FROM readings

┌─id─┬────────────────time─┬──temp─┬─_topic───┬─_offset─┬─_partition─┐
│  1 │ 2020-05-16 23:55:44 │ 14.20 │ readings │       0 │          5 │
│  2 │ 2020-05-16 23:55:45 │ 20.10 │ readings │       1 │          5 │
│  3 │ 2020-05-16 23:55:51 │ 12.90 │ readings │       2 │          5 │
└────┴─────────────────────┴───────┴──────────┴─────────┴────────────┘

顺便说一句，上述过程与在消息格式更改时升级架构的方式相同。此外，物化视图提供了一种非常通用的方法，可以使 Kafka 消息适应目标表行。您甚至可以定义多个具体化视图，以将消息流拆分到不同的目标表中。

从 ClickHouse 写入 Kafka

在本教程的最后，我们将展示如何将消息从 ClickHouse 写回 Kafka。这是一个相对较新的功能，在当前的 Altinity 稳定版本 19.16.18.85 中可用。

让我们首先在 Kafka 中创建一个新主题来包含消息。我们称其为“readings_high”

kafka-topics \
--bootstrap-server kafka:9092 \
--topic readings_high \
--create --partitions 6 \
--replication-factor 2

接下来，我们需要使用 Kafka 表引擎定义一个指向新主题的表。事实证明，此表可以读取和写入消息，但在此示例中，我们将仅使用它进行写入。

CREATE TABLE readings_high_queue (
    readings_id Int32,
    time DateTime,
    temperature Decimal(5,2)
)
ENGINE = Kafka
SETTINGS kafka_broker_list = 'kafka:9092',
         kafka_topic_list = 'readings_high',
         kafka_group_name = 'readings_high_consumer_group1',
         kafka_format = 'CSV',
         kafka_max_block_size = 1048576;

最后，让我们添加一个实例化视图，将温度大于 20.0 的所有行传输到 readings_high_queue 表。此示例说明了 ClickHouse 物化视图的另一个用例，即在特定条件下生成事件。

CREATE MATERIALIZED VIEW readings_high_queue_mv TO readings_high_queue AS
SELECT readings_id, time, temperature FROM readings
WHERE toFloat32(temperature) >= 20.0

在单独的终端窗口中启动消费者，以从 Kafka 上的 readings_high 主题打印消息，如下所示。这将允许您在 ClickHouse 将行写入 Kafka 时查看行。

kafka-console-consumer --bootstrap-server kafka:9092 --topic readings_high

最后，加载一些数据，这些数据将演示如何写回 Kafka。让我们在原始主题中添加一个新批量。在另一个窗口中运行以下命令。

kafka-console-producer --broker-list kafka:9092 --topic readings <<END
4,"2020-05-16 23:55:52",9.7
5,"2020-05-16 23:55:56",25.3
6,"2020-05-16 23:55:58",14.1
END

几秒钟后，您将在运行 kafka-console-consumer 命令的窗口中看到第二行弹出。它应如下所示：

5,"2020-05-16 23:55:56",25.3

故障处理

如果您在使用任何示例时遇到问题，请查看 ClickHouse 日志。如果尚未启用跟踪日志记录，请启用跟踪日志记录。您可以看到如下消息，这些消息表示 Kafka 表引擎中的活动。

2020.05.17 07:24:20.609147 [ 64 ] {} <Debug> StorageKafka (readings_queue): Started streaming to 1 attached views

所有错误将保存在在clickhouse-server.err.log中。

结论

正如这篇博客文章所展示的，Kafka 表引擎提供了一种简单而强大的方法来集成 Kafka 主题和 ClickHouse 表。显然，管理集成还有很多工作要做，尤其是在生产系统中。我们希望本文能帮助您入门，并使您能够自己探索其他可能性。

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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

数据库和数据仓库的区别

数据库和数据仓库的区别

数据仓库是在数据库已知大量存在的前提下&#xff0c;为了进一步挖掘数据资源&#xff0c;为了决策需要产生的&#xff1b;数据仓库在设计的时候有意添加反范式设计&#xff0c;目的是提高查询效率 对比内容数据库数据仓库数据内容近期值历史的 归档的数据数据目标面向业务操作…
阅读更多...
【数学建模】《实战数学建模：例题与讲解》第十三讲-相关分析（含Matlab代码）

【数学建模】《实战数学建模：例题与讲解》第十三讲-相关分析（含Matlab代码）

【数学建模】《实战数学建模&#xff1a;例题与讲解》第十三讲-相关分析&#xff08;含Matlab代码&#xff09; 基本概念典型相关分析综合评价模型对应分析因子分析聚类分析 习题10.41. 题目要求2.解题过程3.程序 习题10.51. 题目要求2.解题过程3.程序 习题10.6&#xff08;1&a…
阅读更多...
[原创][R语言]股票分析实战:周级别涨幅趋势的相关性

[原创][R语言]股票分析实战:周级别涨幅趋势的相关性

[简介]常用网名: 猪头三 出生日期: 1981.XX.XX QQ联系: 643439947 个人网站: 80x86汇编小站 https://www.x86asm.org 编程生涯: 2001年~至今[共22年] 职业生涯: 20年 开发语言: C/C、80x86ASM、PHP、Perl、Objective-C、Object Pascal、C#、Python 开发工具: Visual Studio、De…
阅读更多...
SQL Server数据库使用T-SQL语句简单填充

SQL Server数据库使用T-SQL语句简单填充

文章目录 操作步骤&#xff1a;1.新建数据库起名RGB2.新建表起名rgb3.添加三个列名4.点击新建查询5.填入以下T-SQL语句&#xff0c;点击执行&#xff08;F5&#xff09;6.刷新之后&#xff0c;查看数据 操作环境&#xff1a; win10 Microsoft SQL Server Management Studio 20…
阅读更多...
Leetcode—237.删除链表中的节点【中等】

Leetcode—237.删除链表中的节点【中等】

2023每日刷题&#xff08;六十&#xff09; Leetcode—237.删除链表中的节点 偷天换日实现代码 /*** Definition for singly-linked list.* struct ListNode {* int val;* ListNode *next;* ListNode(int x) : val(x), next(NULL) {}* };*/ class Solution { pub…
阅读更多...
动态内存管理，malloc和calloc以及realloc函数用法

动态内存管理，malloc和calloc以及realloc函数用法

目录 一.malloc函数的介绍 malloc的用法 举个例子 注意点 浅谈数据结构里的动态分配空间 二.calloc函数的介绍 三.realloc函数的介绍 四.柔性数组的介绍 为什么有些时候动态内存函数头文件是malloc.h,有些时候却是stdlib.h 一.malloc函数的介绍 malloc其实就是动态开辟…
阅读更多...
npm ,yarn 更换使用国内镜像源，阿里源，清华大学源

npm ,yarn 更换使用国内镜像源，阿里源，清华大学源

在平时开发当中&#xff0c;我们经常会使用 Npm&#xff0c;yarn 来构建 web 项目。但是npm默认的源的服务器是在国外的&#xff0c;如果没有梯子的话。会感觉特别特别慢&#xff0c;所以&#xff0c;使用国内的源是非常有必要的。 在这里插入图片描述 Nnpm&#xff0c; yarn …
阅读更多...
[极客大挑战 2019]BuyFlag1

[极客大挑战 2019]BuyFlag1

打开网站&#xff1a; 右上角有个菜单 (menu) &#xff0c;先点一下&#xff0c;然后就进入了 pay.php 页面。 其中关键信息如下&#xff1a; ## FlagFlag need your 100000000 money### attentionIf you want to buy the FLAG:You must be a student from CUIT!!!You must…
阅读更多...
云端赋能大湾区：华为云照亮数字化转型之路

云端赋能大湾区：华为云照亮数字化转型之路

编辑&#xff1a;阿冒 设计&#xff1a;沐由 在中国的经济版图上&#xff0c;大湾区是极其重要的增长引擎。这块富有活力和创新力的经济区域里&#xff0c;荟聚了大量的高新技术企业&#xff0c;以及一批创新孵化器和科研机构&#xff0c;产业升级和技术创新的氛围格外浓烈。 1…
阅读更多...
「Verilog学习笔记」单端口RAM

「Verilog学习笔记」单端口RAM

专栏前言 本专栏的内容主要是记录本人学习Verilog过程中的一些知识点&#xff0c;刷题网站用的是牛客网 timescale 1ns/1nsmodule RAM_1port(input clk,input rst,input enb,input [6:0]addr,input [3:0]w_data,output wire [3:0]r_data ); //*************code***********//re…
阅读更多...
python实现websocket上传音频并测试

python实现websocket上传音频并测试

WebSocket是一种全双工通信协议&#xff0c;允许在单个TCP连接上进行双向通信。WebSocket协议允许服务器通过将请求头Upgrade设置为WebSocket来升级HTTP连接。这使得WebSocket协议可以在浏览器和服务器之间建立持久连接&#xff0c;能够实现实时数据传输和通信。 WebSocket协议…
阅读更多...
C++STL中string详解（零基础/小白，字符串）

C++STL中string详解（零基础/小白，字符串）

目录 1. 基本概念&#xff1a; 1.1 本质&#xff1a; 1.2 string和char*区别&#xff1a; 1.3 特点&#xff1a; 2. 构造函数(初始化) 3. 赋值操作 4. 字符串拼接 5 查找 和 替换 6. 字符串比较 7. 字符存取 8. 插入和删除 9. 子串获取 1. 基本概念&#xff1a; 1.…
阅读更多...
Pytorch从零开始实战13

Pytorch从零开始实战13

Pytorch从零开始实战——ResNet与DenseNet探索 本系列来源于365天深度学习训练营 原作者K同学 文章目录 Pytorch从零开始实战——ResNet与DenseNet探索环境准备数据集模型选择开始训练可视化总结 环境准备 本文基于Jupyter notebook&#xff0c;使用Python3.8&#xff0c;P…
阅读更多...
搭建你的知识付费小程序平台：源码解析与技术实现

搭建你的知识付费小程序平台：源码解析与技术实现

知识付费小程序平台在当今数字化时代扮演着越来越重要的角色&#xff0c;为教育者和学习者提供了一个灵活、便捷的学习环境。本文将以关键词“知识付费小程序源码”为基础&#xff0c;探讨如何搭建一个功能强大的知识付费小程序平台&#xff0c;并提供一些基础的技术代码示例。…
阅读更多...
iClient3D 图元操作

iClient3D 图元操作

1. S3MTilesLayer&#xff0c;S3M(Spatial 3D Model)图层类 S3MTilesLayer&#xff0c;S3M(Spatial 3D Model)图层类&#xff0c;通过该图层实现加载三维切片缓存&#xff0c;包括倾斜摄影模型、BIM模型、点云数据、精细模型、矢量数据、符号等。 那S3MTilesLayer中针对图元的…
阅读更多...
vue2 tailwindcss jit模式下热更新失效

vue2 tailwindcss jit模式下热更新失效

按照网上教程安装的tailwindcss&#xff0c;但是修改类名后热更新的时候样式没有生效&#xff0c;参考了大佬的文章&#xff0c;解决了该问题。 安装cross-env 修改前 "dev": " vue-cli-service serve", 修改后 "dev": "cross-env TAILWIN…
阅读更多...
MYSQl基础操作命令合集与详解

MYSQl基础操作命令合集与详解

MySQL入门 先来个总结 SQL语言分类 DDL&#xff08;Data Definition Language&#xff09; - 数据定义语言: 用于定义和管理数据库结构&#xff0c;包括创建、修改和删除数据库对象。 示例&#xff1a;CREATE, ALTER, DROP等语句。 DML&#xff08;Data Manipulation Lan…
阅读更多...
Axios入门案例——后端学习

Axios入门案例——后端学习

目录 后端准备 导入依赖 解决跨域 User实体类 DemoController测试接口 前端准备 项目结构 axios.js axios.html 开始测试 后端结果 前端结果 后端准备 导入依赖 案例会用到以下的三个依赖。 <dependency><groupId>org.springframework.boot</gro…
阅读更多...
nginx反向代理实践指南：访问Tomcat

nginx反向代理实践指南：访问Tomcat

目录 前言1 实现的效果2 访问流程分析3 安装tomcat并测试4 配置4.1 在Windows系统的hosts文件进行域名和IP对应关系的配置4.2 在NGINX进行请求转发的配置&#xff08;反向代理配置&#xff09; 5 最终测试结论 前言 从Windows系统访问Tomcat Web应用程序&#xff0c;设置和配置…
阅读更多...
mybatisplus使用雪花id通过swagger返回ID时精度丢失问题

mybatisplus使用雪花id通过swagger返回ID时精度丢失问题

在使用mybatisplus自带雪花的时候会发现返回的ID是19位的长度&#xff0c;因此在通过swagger页面展示的时候会发现后端返回的和页面展示的ID不一致问题。是因为精度丢失的问题。因此需要更改雪花ID的长度跟踪进去&#xff1a;发现是DefaultIdentifierGenerator类实现了Identifi…
阅读更多...
最新文章

Copyright @ 2022~2023