ClickHouse 与 Kafka 集成：实现高吞吐实时数据管道

/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}

ClickHouse 与 Kafka 集成：实现高吞吐实时数据管道

ClickHouse 作为列式数据库管理系统，以其卓越的查询性能和实时数据处理能力著称。而 Kafka 作为分布式流处理平台，提供了高吞吐、可持久化的消息队列服务。将两者结合，可以构建强大的实时数据分析管道，满足现代数据应用对实时性、高可用性和可扩展性的需求。

ClickHouse 提供了 Kafka 引擎，允许直接从 Kafka 主题中读取数据。同时，通过物化视图管道，可以实现数据的实时处理和转换，将原始数据转化为分析友好的格式。这种架构特别适用于日志分析、IoT 数据处理、实时监控等场景。

1.1 集成优势

• 高吞吐量：ClickHouse 能够处理每秒数百万行数据，Kafka 提供高吞吐的消息传递，两者结合可实现毫秒级数据处理
• 实时性：数据从生产到消费、处理的端到端延迟可达毫秒级
• 可扩展性：Kafka 和 ClickHouse 均支持水平扩展，可根据数据增长灵活调整资源
• 容错性：Kafka 提供持久化存储，ClickHouse 支持数据副本，确保系统可靠性

2. Kafka 引擎配置与使用

ClickHouse 通过 Kafka 引擎实现与 Kafka 集成的核心功能。Kafka 引擎允许 ClickHouse 直接从 Kafka 主题中读取数据，而无需额外的 ETL 流程。

2.1 Kafka 引擎创建

首先，需要在 ClickHouse 中创建使用 Kafka 引擎的表：

CREATE TABLE kafka_engine_table (
timestamp DateTime,
user_id String,
event_type String,
value Float32
) ENGINE = Kafka('kafka-broker1:9092,kafka-broker2:9092,kafka-broker3:9092',
'events_topic',
'clickhouse_consumer_group',
'JSONEachRow');

参数说明：

• 第一个参数：Kafka 集群地址，多个地址用逗号分隔
• 第二个参数：要消费的 Kafka 主题名称
• 第三个参数：消费者组名称
• 第四个参数：消息格式，支持 'JSONEachRow'、'TabSeparated'、'CSV' 等多种格式

2.2 消息格式处理

Kafka 消息格式直接影响 ClickHouse 的数据解析效率。以下是几种常见格式的处理方式：

JSON 格式：

CREATE TABLE json_events (
timestamp DateTime,
user_id String,
event_type String,
value Float32
) ENGINE = Kafka('kafka-brokers:9092', 'events_topic', 'group1', 'JSONEachRow');

CSV 格式：

CREATE TABLE csv_events (
timestamp DateTime,
user_id String,
event_type String,
value Float32
) ENGINE = Kafka('kafka-brokers:9092', 'events_topic', 'group1', 'CSV');

自定义格式：

对于复杂的消息格式，可以使用 ClickHouse 的自定义格式处理器或预处理脚本将数据转换为标准格式。

2.3 消费者管理

ClickHouse 作为 Kafka 消费者，支持以下管理操作：

— 查看当前消费的偏移量
SELECT * FROM system.kafka_consumer_offsets WHERE database = currentDatabase();
— 手动重置偏移量（谨慎使用）
ALTER TABLE kafka_engine_table MODIFY SETTING kafka_reset_offset = 'earliest';

3. 物化视图管道实现

物化视图是 ClickHouse 处理实时数据的强大功能，它与 Kafka 引擎结合可以实现数据转换、聚合和实时写入结果表。

3.1 基本物化视图创建

首先创建目标表用于存储处理后的数据：

CREATE TABLE processed_events (
date Date,
hour UInt8,
user_id String,
event_type String,
count UInt64,
total_value Float64,
avg_value Float64
) ENGINE = MergeTree()
PARTITION BY date
ORDER BY (hour, user_id, event_type);

然后创建物化视图，实现从 Kafka 引擎表到目标表的数据转换：

CREATE MATERIALIZED VIEW mv_processed_events
ENGINE = SummingMergeTree()
PARTITION BY date
ORDER BY (hour, user_id, event_type)
AS SELECT
toDate(timestamp) AS date,
toHour(timestamp) AS hour,
user_id,
event_type,
count() AS count,
sum(value) AS total_value,
avg(value) AS avg_value
FROM kafka_engine_table
GROUP BY date, hour, user_id, event_type;

3.2 多级处理管道

对于复杂的数据处理场景，可以构建多级物化视图管道：

— 第一级：原始数据预处理
CREATE TABLE raw_events (
timestamp DateTime,
user_id String,
event_type String,
value Float32,
raw_data String
) ENGINE = Kafka('kafka-brokers:9092', 'raw_events_topic', 'group2', 'JSONEachRow');
— 第二级：数据清洗和转换
CREATE MATERIALIZED VIEW mv_cleaned_events
ENGINE = MergeTree()
PARTITION BY toDate(timestamp)
ORDER BY (user_id, event_type)
AS SELECT
timestamp,
user_id,
event_type,
value,
— 数据清洗逻辑，如过滤异常值
if(value > 0 AND value < 1000000, value, NULL) AS cleaned_value,
— 数据转换
lower(event_type) AS normalized_event_type
FROM raw_events
WHERE timestamp > now() – INTERVAL 1 DAY;
— 第三级：聚合分析
CREATE MATERIALIZED VIEW mv_aggregated_events
ENGINE = SummingMergeTree()
PARTITION BY toDate(timestamp)
ORDER BY (user_id, normalized_event_type)
AS SELECT
toDate(timestamp) AS date,
user_id,
normalized_event_type,
count() AS count,
sum(cleaned_value) AS total_value,
avg(cleaned_value) AS avg_value
FROM mv_cleaned_events
GROUP BY date, user_id, normalized_event_type;

3.3 异步物化视图

ClickHouse 22.11 版本引入了异步物化视图，显著提高了大规模数据处理场景的性能：

— 创建异步物化视图
CREATE MATERIALIZED VIEW mv_async_events
TO processed_events
AS SELECT
toDate(timestamp) AS date,
toHour(timestamp) AS hour,
user_id,
event_type,
count() AS count,
sum(value) AS total_value
FROM kafka_engine_table
GROUP BY date, hour, user_id, event_type;
— 查看异步物化视图状态
SELECT * FROM system.materialized_views WHERE database = currentDatabase();

异步物化视图的优势：

• 降低主查询负载
• 提高整体吞吐量
• 支持更高的并发写入

4. 实时写入最佳实践

为了充分发挥 ClickHouse 与 Kafka 集成的威力，需要遵循以下最佳实践：

4.1 生产者优化

Kafka 生产者的配置直接影响数据写入性能：

# 生产者配置示例
bootstrap.servers=kafka-broker1:9092,kafka-broker2:9092,kafka-broker3:9092
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
batch.size=16384
linger.ms=5
buffer.memory=33554432
max.request.size=1048576
compression.type=snappy
acks=1

优化要点：

• 批处理：适当增大批处理大小（batch.size）减少网络往返
• 压缩：使用 Snappy 或 LZ4 压缩减少网络传输量
• 确认机制：根据业务需求设置合适的 acks（0/1/all）
• 内存缓冲：合理配置缓冲区大小避免内存不足

4.2 ClickHouse 写入优化

对于 ClickHouse 表的写入，可采用以下优化策略：

表引擎选择：

— 对于高频写入场景，使用 Log 引擎作为缓冲
CREATE TABLE events_buffer (
timestamp DateTime,
user_id String,
event_type String,
value Float32
) ENGINE = Log()
PARTITION BY toDate(timestamp);
— 数据定期从缓冲表聚合到目标表
CREATE TABLE events_aggregated (
date Date,
hour UInt8,
user_id String,
event_type String,
count UInt64,
total_value Float64
) ENGINE = SummingMergeTree()
PARTITION BY date
ORDER BY (date, hour, user_id, event_type);
— 定期执行聚合
INSERT INTO events_aggregated
SELECT
toDate(timestamp) AS date,
toHour(timestamp) AS hour,
user_id,
event_type,
count() AS count,
sum(value) AS total_value
FROM events_buffer
GROUP BY date, hour, user_id, event_type;

批量写入：

— 使用 INSERT INTO … SELECT 实现批量写入
INSERT INTO processed_events
SELECT
toDate(timestamp) AS date,
toHour(timestamp) AS hour,
user_id,
event_type,
count() AS count,
sum(value) AS total_value
FROM kafka_engine_table
WHERE timestamp > now() – INTERVAL 1 HOUR
GROUP BY date, hour, user_id, event_type;

分区和排序键优化：

— 根据查询模式优化分区和排序键
CREATE TABLE optimized_events (
date Date,
event_type String,
user_id String,
value Float32,
attributes String
) ENGINE = MergeTree()
PARTITION BY date
ORDER BY (event_type, user_id, value)
SETTINGS index_granularity = 8192;

4.3 监控与调优

建立完善的监控机制对系统性能至关重要：

关键监控指标：

• Kafka 消费延迟
• ClickHouse 查询响应时间
• 系统资源使用率（CPU、内存、磁盘 I/O）
• 表数据增长趋势

调优手段：

• 调整合并树参数（如 index_granularity）
• 优化查询语句
• 根据负载情况扩展 ClickHouse 集群
• 调整 Kafka 分区数量与副本因子

5. 最小示例与注意事项

5.1 最小示例

以下是一个完整的 ClickHouse 与 Kafka 集成最小示例：

# 1. 启动 Kafka 主题
kafka-topics.sh –create –topic clickhouse_demo \\
–bootstrap-server kafka-broker:9092 \\
–partitions 3 –replication-factor 2
# 2. 生产者发送数据（JSON 格式）
echo '{"timestamp":"2023-01-01 12:00:00", "user_id":"user1", "event_type":"click", "value":1.5}' | \\
kafka-console-producer.sh –topic clickhouse_demo \\
–bootstrap-server kafka-broker:9092
# 3. ClickHouse 创建 Kafka 表
CREATE TABLE kafka_demo (
timestamp DateTime,
user_id String,
event_type String,
value Float32
) ENGINE = Kafka('kafka-broker:9092', 'clickhouse_demo', 'demo_group', 'JSONEachRow');
# 4. 创建处理后的目标表
CREATE TABLE processed_demo (
date Date,
user_id String,
event_type String,
count UInt64,
total_value Float64
) ENGINE = MergeTree()
PARTITION BY date
ORDER BY (user_id, event_type);
# 5. 创建物化视图
CREATE MATERIALIZED VIEW mv_processed_demo
ENGINE = SummingMergeTree()
PARTITION BY date
ORDER BY (user_id, event_type)
AS SELECT
toDate(timestamp) AS date,
user_id,
event_type,
count() AS count,
sum(value) AS total_value
FROM kafka_demo
GROUP BY date, user_id, event_type;
# 6. 查询结果
SELECT * FROM processed_demo ORDER BY date, user_id, event_type;

5.2 注意事项

通过遵循以上实践，可以构建稳定高效、高吞吐量的 ClickHouse 与 Kafka 集成系统，满足实时数据分析的需求。

#publish-mermaid-0-1789611650522{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#publish-mermaid-0-1789611650522 .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#publish-mermaid-0-1789611650522 .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#publish-mermaid-0-1789611650522 .error-icon{fill:#552222;}#publish-mermaid-0-1789611650522 .error-text{fill:#552222;stroke:#552222;}#publish-mermaid-0-1789611650522 .edge-thickness-normal{stroke-width:1px;}#publish-mermaid-0-1789611650522 .edge-thickness-thick{stroke-width:3.5px;}#publish-mermaid-0-1789611650522 .edge-pattern-solid{stroke-dasharray:0;}#publish-mermaid-0-1789611650522 .edge-thickness-invisible{stroke-width:0;fill:none;}#publish-mermaid-0-1789611650522 .edge-pattern-dashed{stroke-dasharray:3;}#publish-mermaid-0-1789611650522 .edge-pattern-dotted{stroke-dasharray:2;}#publish-mermaid-0-1789611650522 .marker{fill:#333333;stroke:#333333;}#publish-mermaid-0-1789611650522 .marker.cross{stroke:#333333;}#publish-mermaid-0-1789611650522 svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#publish-mermaid-0-1789611650522 p{margin:0;}#publish-mermaid-0-1789611650522 .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#publish-mermaid-0-1789611650522 .cluster-label text{fill:#333;}#publish-mermaid-0-1789611650522 .cluster-label span{color:#333;}#publish-mermaid-0-1789611650522 .cluster-label span p{background-color:transparent;}#publish-mermaid-0-1789611650522 .label text,#publish-mermaid-0-1789611650522 span{fill:#333;color:#333;}#publish-mermaid-0-1789611650522 .node rect,#publish-mermaid-0-1789611650522 .node circle,#publish-mermaid-0-1789611650522 .node ellipse,#publish-mermaid-0-1789611650522 .node polygon,#publish-mermaid-0-1789611650522 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#publish-mermaid-0-1789611650522 .rough-node .label text,#publish-mermaid-0-1789611650522 .node .label text,#publish-mermaid-0-1789611650522 .image-shape .label,#publish-mermaid-0-1789611650522 .icon-shape .label{text-anchor:middle;}#publish-mermaid-0-1789611650522 .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#publish-mermaid-0-1789611650522 .rough-node .label,#publish-mermaid-0-1789611650522 .node .label,#publish-mermaid-0-1789611650522 .image-shape .label,#publish-mermaid-0-1789611650522 .icon-shape .label{text-align:center;}#publish-mermaid-0-1789611650522 .node.clickable{cursor:pointer;}#publish-mermaid-0-1789611650522 .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#publish-mermaid-0-1789611650522 .arrowheadPath{fill:#333333;}#publish-mermaid-0-1789611650522 .edgePath .path{stroke:#333333;stroke-width:1px;}#publish-mermaid-0-1789611650522 .flowchart-link{stroke:#333333;fill:none;}#publish-mermaid-0-1789611650522 .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#publish-mermaid-0-1789611650522 .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#publish-mermaid-0-1789611650522 .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#publish-mermaid-0-1789611650522 .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#publish-mermaid-0-1789611650522 .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#publish-mermaid-0-1789611650522 .cluster text{fill:#333;}#publish-mermaid-0-1789611650522 .cluster span{color:#333;}#publish-mermaid-0-1789611650522 div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#publish-mermaid-0-1789611650522 .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#publish-mermaid-0-1789611650522 rect.text{fill:none;stroke-width:0;}#publish-mermaid-0-1789611650522 .icon-shape,#publish-mermaid-0-1789611650522 .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#publish-mermaid-0-1789611650522 .icon-shape p,#publish-mermaid-0-1789611650522 .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#publish-mermaid-0-1789611650522 .icon-shape .label rect,#publish-mermaid-0-1789611650522 .image-shape .label rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#publish-mermaid-0-1789611650522 .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#publish-mermaid-0-1789611650522 .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#publish-mermaid-0-1789611650522 .node .neo-node{stroke:#9370DB;}#publish-mermaid-0-1789611650522 [data-look=\”neo\”].node rect,#publish-mermaid-0-1789611650522 [data-look=\”neo\”].cluster rect,#publish-mermaid-0-1789611650522 [data-look=\”neo\”].node polygon{stroke:#9370DB;filter:drop-shadow(1px 2px 2px rgba(185, 185, 185, 1));}#publish-mermaid-0-1789611650522 [data-look=\”neo\”].swimlane.cluster rect{filter:none;}#publish-mermaid-0-1789611650522 [data-look=\”neo\”].node path{stroke:#9370DB;stroke-width:1px;}#publish-mermaid-0-1789611650522 [data-look=\”neo\”].node .outer-path{filter:drop-shadow(1px 2px 2px rgba(185, 185, 185, 1));}#publish-mermaid-0-1789611650522 [data-look=\”neo\”].node .neo-line path{stroke:#9370DB;filter:none;}#publish-mermaid-0-1789611650522 [data-look=\”neo\”].node circle{stroke:#9370DB;filter:drop-shadow(1px 2px 2px rgba(185, 185, 185, 1));}#publish-mermaid-0-1789611650522 [data-look=\”neo\”].node circle .state-start{fill:#000000;}#publish-mermaid-0-1789611650522 [data-look=\”neo\”].icon-shape .icon{fill:#9370DB;filter:drop-shadow(1px 2px 2px rgba(185, 185, 185, 1));}#publish-mermaid-0-1789611650522 [data-look=\”neo\”].icon-shape .icon-neo path{stroke:#9370DB;filter:drop-shadow(1px 2px 2px rgba(185, 185, 185, 1));}#publish-mermaid-0-1789611650522 :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

生产者生成数据

Kafka 集群

ClickHouse Kafka 引擎

ClickHouse 表

物化视图处理

结果表