面向车联网实时链路的 ClickHouse 全栈教程——第 18 章 全链路整合：ClickHouse + Paimon + Flink + Doris 生产部署实战

第 18 章 全链路整合：ClickHouse + Paimon + Flink + Doris 生产部署实战

章节定位：第 1 至 17 章带你把 ClickHouse 单产品啃透了——但生产环境不会只有 ClickHouse 一个组件。一辆车的遥测数据从车载 T-Box 出来 → Kafka → Flink ODS 做清洗 → 一份写 Paimon 湖仓一份写 ClickHouse → ClickHouse 做多 shard 聚合 → Doris 读 ClickHouse 做实时看板。这一章把第 16、17 章的集群运维、复制、分布式查询和数据栈里的其他四个核心组件（Flink / Paimon / Kafka / Doris）整合起来，讲清每个组件的角色边界、数据流动方向、以及"让全链路跑起来"的工程细节。核心心智：ClickHouse 是查询和聚合层，Paimon 是原始数据层，Flink 是流处理层，Kafka 是解耦管道，Doris 是BI 看板层——五者各司其职，不互相取代，但边界模糊处（“为什么不直接 Doris 写 Kafka？”、“为什么 ClickHouse 不直接读 Iceberg 然后跳过 Paimon？”）就是你作为架构师要做的决策。

学习目标

读完本章，你应该能够：

• 画出完整的车辆数据全链路架构图，标注每个组件的角色边界和数据流动方向
• 说出 Flink → ClickHouse 三种写入方式（Flink Connector / JDBC Sink / Flink 写 Kafka → ClickHouse Kafka 引擎）的优缺点和适用场景
• 用 Flink Connector 给 demo.vehicle_telemetry 写 Exactly-Once 语义的批量写入，说出关键参数（maxBatchSize / maxInFlightRequests / flushInterval）
• 用 ClickHouse 的 PaimonS3 / IcebergS3 / HudiS3 / Hive 外表引擎读 Paimon 湖仓数据，知道它们都是只读外表
• 给外表建本地 Materialized View 管道，把湖仓数据周期性物化到 ClickHouse 本地表做高效查询
• 用 Doris 的 JDBC Catalog 连 ClickHouse，说出 CREATE CATALOG 语法 + ClickHouse JDBC 驱动配置
• 说出 ClickHouse 读 Doris 的反向方式（ClickHouse 的 JDBC 引擎），以及什么时候选 Doris 写 ClickHouse、什么时候选 ClickHouse 写 Doris
• 用 ClickHouse 的 Kafka 引擎（Kafka table + Materialized View 管道）做流式订阅，说出 Kafka 引擎的 commit 机制（kafka_commit_on_select）
• 说清 ClickHouse Cloud 用户跳过哪些节（Kafka 引擎、冷热分层、自建 Keeper；用 ClickPipes 替代 Flink connector 直写）

18.1 全链路架构总览

18.1.1 五个组件的角色边界

我们的车辆数据全链路涉及五个核心组件，每个有明确的角色分工：

18.1.2 完整数据流：从车载传感器到 BI 看板

车辆 T-Box（100 万辆，每秒 ~100 万条原始遥测）
│
│ Kafka producer（带 VIN 分区 + ack=all）
▼
┌──────────────────────────────┐
│ Kafka 集群 │
│ topic: vehicle_raw │
│ (replication-factor=3) │
└──────┬───────────┬───────────┘
│ │
│ │ consumer group "paimon-ods"
│ ▼
│ ┌────────────────────────────┐
│ │ Flink Job: ODS 清洗 │
│ │ 1. 脏数据过滤（speed<0等） │
│ │ 2. 字段补全（城市名称映射） │
│ │ 3. 格式统一（时间戳毫秒→秒） │
│ └──────┬───────────┬──────────┘
│ │ │
│ │ sink: Paimon S3 │ sink: ClickHouse Connector
│ ▼ ▼
│ ┌──────────────┐ ┌──────────────────────┐
│ │ Apache Paimon │ │ ClickHouse 多 shard │
│ │ S3 / HDFS │ │ 集群 │
│ │ │ │ ReplicatedMergeTree │
│ │ 原始历史数据 │ │ 近实时聚合层 │
│ │ (只读给下游) │ │ │
│ └──────┬───────┘ └──────┬───────────────┘
│ │ │
│ │ PaimonS3 外表 │ JDBC 引擎
│ ▼ ▼
│ ┌────────────────────────────────────────────┐
│ │ ClickHouse │
│ │ 外表 + Materialized View → 本地表 │
│ │ (湖仓数据物化 + 本地聚合) │
│ └──────────────────────┬─────────────────────┘
│ │
│ │ Doris JDBC Catalog
│ ▼
│ ┌──────────────────────────────────────────┐
│ │ Apache Doris │
│ │ CREATE CATALOG clickhouse_federated │
│ │ SELECT * FROM clickhouse_federated. │
│ │ demo.vehicle_telemetry WHERE … │
│ │ │
│ │ BI 看板（Superset / Grafana / 自研） │
│ └──────────────────────────────────────────┘
│
└─ 备选路径：Flink 直接写 ClickHouse Kafka 引擎
（Flink → Kafka topic vehicle_cleaned → ClickHouse
Kafka 表 + Materialized View 管道）

图 18-1 车辆数据全链路五组件架构

注意图里的两个核心数据流：

• 实时路径（粗线）：Kafka → Flink → ClickHouse 多 shard → Doris 看板 → 用户实时查询。延迟目标 < 30 秒。
• 湖仓路径（虚线）：Kafka → Flink → Paimon S3 → ClickHouse PaimonS3 外表 → 本地物化视图 → 长期趋势分析。这是历史数据回补 + 冷数据分析的路径，延迟目标 天级。

18.1.3 为什么不简化：每个组件都是必要的

18.1.4 ClickHouse 在全链路中的"枢纽"地位

把图 18-1 的 ClickHouse 单独拉出来看，它是整个链路的枢纽节点——上游有三个写入口、下游有两个读出口：

上游（三条写入口）
┌──────────────┬──────────────┐
▼ ▼ ▼
Flink Connector Kafka 引擎 PaimonS3
(实时批量写入) (流式订阅) (外表只读)
│ │ │
│ │ │
└──────────────┼──────────────┘
▼
┌───────────────────┐
│ ClickHouse 集群 │
│ (4 shard × 3 replica)│
│ │
│ 本地 Replicated │ Materialized View
│ MergeTree 表 │ 管道（外表→本地表）
└────────┬──────────┘
│
┌────────────┼────────────┐
▼ ▼ ▼
下游（两个读出口 + 内部分布式查询）
Doris JDBC 应用层直接 ClickHouse
Catalog 查单机表 Distributed
跨 shard 查

图 18-2 ClickHouse 作为枢纽的三条写入口 + 两个读出口

18.2 Flink → ClickHouse：三种写入方式

Flink 往 ClickHouse 写数据有三条路。选择依据是你的数据特征（批量 vs 流式）、一致性要求（Exactly-Once vs At-Least-Once）、和解耦程度（是否需要 Kafka 作为中间缓冲）。

18.2.1 方式 A：官方 Flink Connector（推荐）

Flink Connector 是 ClickHouse 官方维护的 AsyncSink 实现——基于 ClickHouse 原生 HTTP 协议 + Async 批量写入。这是生产首选，因为：

• 支持 Exactly-Once 语义（23.3+ 版本）
• 基于 ClickHouse 原生 JDBC 协议（不是 HTTP JSON 绕一圈）
• AsyncSinkBase 实现（Flink 1.14+ 标准 Async Sink 接口），性能好
• 自动选 Distributed 表的健康 replica + 按 sharding key 切 shard
• 支持 checkpoint 状态持久化 → Flink 重启后从 checkpoint 恢复

Maven 依赖

<!– Flink 2.0+ –>
<dependency>
<groupId>com.clickhouse.flink</groupId>
<artifactId>flink-connector-clickhouse-2.0.0</artifactId>
<version>1.1.1</version>
<classifier>all</classifier>
</dependency>

<!– Flink 1.17 ~ 1.20 –>
<dependency>
<groupId>com.clickhouse.flink</groupId>
<artifactId>flink-connector-clickhouse-1.17</artifactId>
<version>1.1.1</version>
<classifier>all</classifier>
</dependency>

Java 代码骨架

import com.clickhouse.flink.connector.ClickHouseSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;

// 假设 vehicleData 是从 Kafka source 清洗后的 DataStream
DataStream<VehicleTelemetryRecord> vehicleData = ...;

// 关键参数
ClickHouseSink<VehicleTelemetryRecord> sink = ClickHouseSink.<VehicleTelemetryRecord>builder()
// ClickHouse 连接
.withUrl("jdbc:clickhouse://ck-1-1:8123,ck-1-2:8123,ck-1-3:8123")
.withDatabase("demo")
.withTable("vehicle_telemetry") // 如果是 Distributed 表，直接写分布式表名
.withUsername("flink_writer")
.withPassword("xxx")

// 批量控制
.withMaxBatchSize(50000) // 每批 5 万行（生产推荐 1 万~10 万）
.withMaxInFlightRequests(3) // 同时最多 3 个 in-flight 请求（反压控制）
.withFlushInterval(Duration.ofSeconds(5)) // 5 秒强制 flush（即使没凑满 batch）

// 路由 + 一致性
.withShardingKeyExtractor(record -> record.getVin()) // 按 vin 算 shard
.withWriteMode(WriteMode.INSERT) // INSERT / UPSERT 可选
.withFailOnTimeout(true)

// 序列化（避免 Kryo 开销，用原生 POJO ↔ ClickHouse）
.withConverter(VehicleTelemetryRecord::toClickHouseRow)

.build();

vehicleData
.assignTimestampsAndWatermarks(WatermarkStrategy.noWatermarks())
.sinkTo(sink);

关键参数解释

Exactly-Once 配置

ClickHouse Flink Connector 2.0+ 支持 Exactly-Once（基于 Flink 的 Sink V2 + Two-Phase-Commit）。需要：

18.2.2 方式 B：Flink JDBC Sink（通用但性能一般）

Flink 自带 flink-connector-jdbc，理论上可以接任何支持 JDBC 的数据库包括 ClickHouse：

// Flink 自带 JDBC sink
JdbcSink.sink(
"INSERT INTO demo.vehicle_telemetry VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
(ps, record) -> {
ps.setString(1, record.getVin());
ps.setInt(2, record.getCityId());
// … 其他字段
},
new JdbcExecutionOptions.Builder()
.withBatchSize(50000)
.withBatchIntervalMs(5000)
.build(),
new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
.withUrl("jdbc:clickhouse://ck-1-1:8123")
.withDriverName("com.clickhouse.jdbc.ClickHouseDriver")
.withUsername("flink_writer")
.withPassword("xxx")
.build()
);

但 JDBC Sink 的问题：

• 用的是 ClickHouse HTTP 协议 + JSON 格式绕一圈，比原生 AsyncSink 慢 2~5 倍
• 不支持自动选 replica（只能写一个固定地址）
• 不支持 Exactly-Once（最多 At-Least-Once，配合 ReplacingMergeTree 去重可接受）
• 用 Flink 自带 JdbcSink，没有 sharding_key 概念——写 Distributed 表时可能全部打到同一个 shard

什么时候用：你的 Flink job 里已经在同时写 MySQL + PostgreSQL + ClickHouse，想统一用 Flink 自带 JDBC Sink 减少维护成本、且能接受 ClickHouse 写入比官方 Connector 慢。

18.2.3 方式 C：Flink 写 Kafka → ClickHouse Kafka 引擎（解耦流式管道）

这是最解耦的方式——Flink 不直接碰 ClickHouse，而是把清洗后的数据写进 Kafka，然后 ClickHouse 自己作为 Kafka consumer 订阅。好处是"Flink 挂了 ClickHouse 不受影响、ClickHouse 挂了 Flink 不受影响"。

Flink 侧：写 Kafka

// Flink 清洗后的 vehicleData 写 Kafka topic vehicle_cleaned
FlinkKafkaProducer<VehicleTelemetryRecord> kafkaSink = new FlinkKafkaProducer<>(
"vehicle_cleaned", // topic
new VehicleTelemetryRecordSchema(), // serializer（JSON 格式）
kafkaProperties,
FlinkKafkaProducer.Semantic.EXACTLY_ONCE
);

vehicleData.sinkTo(kafkaSink);

ClickHouse 侧：Kafka 表 + Materialized View 管道

ClickHouse 的 Kafka 表引擎 是 ClickHouse 原生实现的 Kafka consumer——它只做"订阅 + 解析"，然后通过 Materialized View 管道 把数据写到目标表：

— 第 1 步：建 Kafka 表（只负责订阅和解析，不存数据）
— 注意：allow_experimental_kafka_storage_engine 在新版 ClickHouse 里可能已移除
— Cloud 用户用 ClickPipes 替代
CREATE TABLE IF NOT EXISTS demo.vehicle_kafka (
vin String,
city_id UInt32,
report_date Date,
reported_at DateTime,
speed Float32,
battery UInt32,
battery_pct UInt8,
vehicle_status LowCardinality(String),
seq UInt64,
is_deleted UInt8,
ext String
) ENGINE = Kafka()
SETTINGS
kafka_broker_list = 'kafka-1:9092,kafka-2:9092,kafka-3:9092',
kafka_topic_list = 'vehicle_cleaned',
kafka_group_name = 'clickhouse_vehicle_consumer',
kafka_format = 'JSONEachRow',
kafka_num_consumers = 4, — 4 个 consumer（多线程加速）
kafka_commit_every_batch = 1, — 每个 batch 就 commit offset
kafka_handle_error_mode = 'stream'; — 错误也写入表（带 _error 列）

— 第 2 步：建目标表（ReplicatedReplacingMergeTree）
CREATE TABLE IF NOT EXISTS demo.vehicle_telemetry
ENGINE = ReplicatedReplacingMergeTree(
'/clickhouse/tables/{shard}/demo/vehicle_telemetry',
'{replica}',
seq
)
PARTITION BY cityHash64(vin) % 8
ORDER BY (city_id, report_date, vin);

— 第 3 步：建 Materialized View 管道（持续把 Kafka 表的数据 INSERT 到目标表）
CREATE MATERIALIZED VIEW IF NOT EXISTS demo.vehicle_kafka_mv
TO demo.vehicle_telemetry AS
SELECT * FROM demo.vehicle_kafka;

Kafka 引擎的关键设置

⚠️ ClickHouse Cloud 用户注意：Kafka 引擎在 Cloud 上可能不可用或受限。Cloud 推荐用 ClickPipes 服务——托管的 Kafka → ClickHouse 管道，自动管理 consumer group / offset / 扩容。

18.2.4 三种方式对比与车辆场景选型

车辆场景选型建议：

• 短期 PoC / MVP：方式 A（Flink Connector）——最快、最少组件、一次跑通端到端。
• 长期生产、高吞吐、需要解耦：方式 A + 方式 C 组合——Flink 同时开 Flink Connector（实时）和 Kafka sink（解耦备份），ClickHouse 侧开 Flink Connector 直写 + Kafka 引擎管道备份。两条路互为灾备。
• 已在生产跑 Flink 写 Kafka：方式 C 直接接上——不用改 Flink，ClickHouse 加一套 Kafka 引擎 + MV 就行。

18.3 ClickHouse 读湖仓：Paimon / Iceberg / Hudi / Hive 外表

18.3.1 统一心智：全部是只读外表

ClickHouse 集成湖仓的方式高度统一：每张外表引擎都是只读的——ClickHouse 不会往 Paimon/Iceberg/Hudi/Hive 里写数据（写数据的事交给 Flink / Spark / Trino），它只负责"能 SELECT 读到这些数据、能在一次查询里 JOIN 本地表 + 外表"。

外表引擎命名规则按存储后端分：

所有外表都需要：allow_experimental_paimon_storage_engine / allow_experimental_hudi_storage_engine 等配置开启（默认关闭）。

18.3.2 Paimon 外表最常用的场景：车辆历史原始数据

假设 Flink 已经把清洗后的车辆遥测写进了 S3 上的 Paimon 表 s3://vehicle-lake/vehicle_telemetry_paimon/：

— 全局开 Paimon 外表功能（只需要开一次）
SET allow_experimental_paimon_storage_engine = 1;

— 生产推荐：用 Named Collection 存 S3 凭证（不要硬编码到 CREATE TABLE 里）
— 在 config.xml 或 config.d/paimon.xml 里加：
— <clickhouse>
— <named_collections>
— <paimon_vehicle>
— <url>s3://vehicle-lake/vehicle_telemetry_paimon/</url>
— <access_key_id>AKIA…</access_key_id>
— <secret_access_key>xxxxxx</secret_access_key>
— </paimon_vehicle>
— </named_collections>
— </clickhouse>

CREATE TABLE demo.vehicle_telemetry_paimon_local
(
vin String,
city_id UInt32,
report_date Date,
reported_at DateTime,
speed Float32,
battery UInt32,
battery_pct UInt8,
vehicle_status LowCardinality(String),
seq UInt64,
is_deleted UInt8,
ext String
)
ENGINE = PaimonS3(paimon_vehicle, 'JSONEachRow');
— 参数 1：Named Collection 名（或完整 URL + key + secret）
— 参数 2：Paimon 底层数据格式（JSONEachRow / Parquet 等）

外表查询：

— 直接 SELECT 外表（ClickHouse 实时从 S3 读 Paimon snapshot）
SELECT
city_id,
avg(speed) AS avg_speed,
avg(battery_pct) AS avg_battery
FROM demo.vehicle_telemetry_paimon_local
WHERE report_date >= today() – 30
GROUP BY city_id;

— 外表 JOIN 本地表（核心威力！）
— 同一查询里同时读 S3 冷数据 + 本地热数据
SELECT
vin,
report_date,
avg(speed) AS avg_speed,
count() AS cnt,
'cold_from_paimon' AS data_source
FROM demo.vehicle_telemetry_paimon_local
WHERE report_date < today() – 7
GROUP BY vin, report_date

UNION ALL

SELECT
vin,
report_date,
avg(speed) AS avg_speed,
count() AS cnt,
'hot_from_local' AS data_source
FROM demo.vehicle_telemetry — 本地 ReplicatedMergeTree
WHERE report_date >= today() – 7
GROUP BY vin, report_date;

外表 + 本地物化视图管道（把湖仓数据周期性拉到本地做高效查询）：

— 建本地目标表（ReplicatedMergeTree，和外表 schema 相同）
CREATE TABLE demo.vehicle_telemetry_archive
ENGINE = ReplicatedReplacingMergeTree(
'/clickhouse/tables/{shard}/demo/vehicle_telemetry_archive',
'{replica}', seq
)
PARTITION BY toYYYYMM(report_date) — 按月分区（大表分区稀疏一点）
ORDER BY (vin, report_date);

— 物化视图：外表 → 本地表
CREATE MATERIALIZED VIEW demo.vehicle_telemetry_archive_mv
TO demo.vehicle_telemetry_archive AS
SELECT * FROM demo.vehicle_telemetry_paimon_local;

— 手动触发一次全量拉取（后台自动跑）
SYSTEM FLUSH LOGS; — 先确保外表数据加载到内存缓存
OPTIMIZE TABLE demo.vehicle_telemetry_archive FINAL;

18.3.3 Iceberg / Hudi / Hive：选择哪个

三个湖仓外表的差异主要在底层元数据管理方式：

车辆场景建议：Paimon 是 Flink 原生湖仓（Flink 1.15+ 官方支持），我们整个数据栈从 Flink 起 → Paimon 是首选。如果已有 Iceberg 或 Hudi 基础设施，ClickHouse 也能无缝接入（外表引擎只读，不影响 Flink/Spark 那边的写入逻辑）。

18.3.4 外表查询性能优化

外表直接读 S3 远没有本地盘快——几个优化技巧：

<local_cache_for_remote_fs>
<enable>true</enable>
<root_dir>/var/lib/clickhouse/cache</root_dir>
<limit_size>21474836480</limit_size> <!– 20GB –>
</local_cache_for_remote_fs>

18.4 ClickHouse ↔ Doris：双向联邦

Doris 和 ClickHouse 都能通过 JDBC 协议互相查询。选择哪个方向取决于"哪一侧是联邦查询的发起者和结果消费者"。

18.4.1 方向 A：Doris 读 ClickHouse（主流方向）

Doris 的 JDBC Catalog 功能让 Doris 能把 ClickHouse 当成一个"外部数据目录"——像查本地 Doris 表一样查 ClickHouse 的表。这是 BI 看板的主流模式（Doris 做 federated query + 本地 cube 预计算 + BI 工具对接）。

步骤 1：准备 JDBC 驱动

Doris 的 FE（Frontend）需要 ClickHouse JDBC 驱动。把 jar 放到 doris/fe/jdbc_drivers/clickhouse/ 目录下：

# 驱动下载地址
# https://github.com/ClickHouse/clickhouse-jdbc/releases
# 推荐 clickhouse-jdbc-0.6.0.jar + clickhouse-jdbc-0.6.0-all.jar

步骤 2：Doris 侧 CREATE CATALOG

— 在 Doris FE 上执行
CREATE CATALOG vehicle_clickhouse_catalog
PROPERTIES (
"type" = "jdbc",
"jdbc_driver_url" = "https://github.com/ClickHouse/clickhouse-jdbc/releases/download/v0.6.0/clickhouse-jdbc-0.6.0-all.jar",
"jdbc_driver_class" = "com.clickhouse.jdbc.ClickHouseDriver",
"jdbc_url" = "jdbc:clickhouse://ck-1-1:8123,ck-1-2:8123,ck-1-3:8123/demo",
"jdbc_user" = "doris_reader",
"jdbc_password" = "xxx",
— ClickHouse JDBC Catalog 特有的参数
"lower_case_table_names" = "1", — Doris 大小写不敏感，ClickHouse 敏感
"table_meta_cache_size" = "10000", — 缓存 10000 张表的 metadata
"include_database_list" = "demo" — 只映射 demo 库（不扫全集群）
);

步骤 3：像查本地表一样查 ClickHouse

— 切到 ClickHouse 联邦 catalog
USE vehicle_clickhouse_catalog;

— 直接查 ClickHouse 表
SELECT
city_id,
avg(speed) AS avg_speed
FROM demo.vehicle_telemetry — 这是 ClickHouse 里的表
WHERE report_date >= today() – 7
GROUP BY city_id;

— Doris 本地表 JOIN ClickHouse 表（联邦 JOIN）
SELECT
v.city_id,
d.city_name,
avg(v.speed) AS avg_speed
FROM demo.vehicle_telemetry v — ClickHouse 表
JOIN doris_demo.city_dict d — Doris 本地字典表
ON v.city_id = d.city_id
WHERE v.report_date >= today() – 7
GROUP BY v.city_id, d.city_name;

注意 Doris JDBC Catalog 的局限性（Doris 官方自己说的）：

• JDBC Catalog 只适合小数据量查询和数据集成——不能加速 ClickHouse 的查询，也不适合一次性拉大量数据
• 高性能场景用 Doris Routine Load 把 ClickHouse 数据持续同步到 Doris 本地表，然后 BI 看板只查 Doris 本地表
• ClickHouse 表里的 LowCardinality / Array / Tuple 等复杂类型，Doris JDBC Catalog 可能需要 cast

18.4.2 方向 B：ClickHouse 读 Doris（反向）

ClickHouse 的 JDBC 引擎 可以查任何支持 JDBC 的数据库包括 Doris：

— ClickHouse 侧建 Doris 外表（只读）
CREATE TABLE demo.doris_city_dict
(
city_id UInt32,
city_name String,
province String
)
ENGINE = JDBC(
'jdbc:mysql://doris-fe:9030/doris_demo', — Doris FE 用 MySQL 协议
'root',
'',
'city_dict'
);

— 用 Doris MySQL 驱动（放在 ClickHouse 服务器的 lib 目录下）
— 驱动名：mysql-connector-java-8.0.xx.jar

— 查询 + JOIN 本地 ClickHouse 表
SELECT
t.city_id,
d.city_name,
avg(t.speed) AS avg_speed
FROM demo.vehicle_telemetry_local t
JOIN demo.doris_city_dict d ON t.city_id = d.city_id
WHERE t.report_date >= today() – 7
GROUP BY t.city_id, d.city_name;

ClickHouse Cloud 用户注意：JDBC 引擎在 Cloud 上可能受限（外部 JDBC 驱动加载）。Cloud 用户优先用方向 A（Doris 读 ClickHouse）。

18.4.3 双向都能做，怎么选

生产推荐：方向 A 为主（Doris 读 ClickHouse 做 BI）+ 方向 B 按需使用（ClickHouse 偶尔查 Doris 的小字典表做 JOIN）。

18.5 Kafka 引擎深入：流式订阅 + Materialized View 管道

Kafka 引擎（方式 C）在 18.2.3 里已经给了基础骨架。这里深入几个生产高频踩坑点。

18.5.1 Kafka 引擎的本质：只订阅解析，不存数据

Kafka 引擎表本身不存任何数据——它是一张"实时视图"，每次 SELECT 都会触发 Kafka poll。数据流向目标表完全依赖 Materialized View 管道：

Kafka topic vehicle_cleaned
│
│ ClickHouse Kafka consumer（在后台持续 poll）
▼
demo.vehicle_kafka（Kafka 引擎表，不存数据）
│
│ Materialized View 管道
│ （在 Kafka consumer 线程里同步 INSERT）
▼
demo.vehicle_telemetry（ReplicatedReplacingMergeTree，存数据）

图 18-3 Kafka 引擎 + MV 管道的数据流向

18.5.2 管道启停控制

MV 管道创建后自动开始。如果需要临时停掉：

— 停掉 MV（底层是 Detach + Kill query）
SYSTEM STOP MERGES demo.vehicle_kafka_mv;
— 或者直接 DROP MATERIALIZED VIEW（同时停管道）
DROP MATERIALIZED VIEW IF EXISTS demo.vehicle_kafka_mv;

— 恢复
SYSTEM START MERGES demo.vehicle_kafka_mv;

18.5.3 Kafka offset 管理

Kafka 引擎表会在 ClickHouse 服务器的 /var/lib/clickhouse/[your_database]/[kafka_table_name] 目录下存一个 __consumer_offsets 文件。不要手动删这个文件——删了等于重置 consumer group，会从 earliest 或 latest 重新开始消费。

关键设置 kafka_commit_on_select：

• 默认 false：MV 管道消费时，commit offset 在后台按 kafka_commit_every_batch 或 flush_interval 触发。推荐保持 false（MV 管道是消费主体、commit 时机由 MV 控制）。
• 设为 true：每次 SELECT 引擎表时才 commit offset。只在手动调试时用（比如想精确控制"消费到这里为止"）。

18.5.4 故障恢复与 Exactly-Once

Kafka 引擎 + MV 管道的语义是 At-Least-Once（不是 Exactly-Once）——原因：

解决办法：目标表用 ReplacingMergeTree（我们的 demo.vehicle_telemetry 就是）——同一个 (vin, seq) 重复插入，merge 阶段自动去重。At-Least-Once + ReplacingMergeTree = 效果等同于 Exactly-Once。

18.5.5 与 Flink Connector 的协同：两条写入路互为灾备

Kafka topic vehicle_cleaned（Flink sink 写入）
│
┌─────┴──────┐
▼ ▼
Kafka 引擎管道 Flink Connector 直写
（异步、解耦） （同步、高性能）
│ │
▼ ▼
demo.vehicle_telemetry（ReplicatedReplacingMergeTree）
同一目标表，两条路都写，ReplacingMergeTree 自动去重

图 18-4 两条写入路互为灾备的架构

当 Flink Connector 直写不可用时（网络问题 / ClickHouse 抖动），Flink 的 Kafka sink 继续往 topic 写，ClickHouse 的 Kafka 引擎管道继续消费 → 写入不中断。反之当 Kafka 集群故障时，Flink Connector 直写继续 → 写入也不中断。

两条路同时正常时 → 同一行数据可能被写两次 → ReplacingMergeTree 按主键去重 → 不影响正确性。

18.6 端到端部署

18.6.1 组件清单与端口

18.6.2 Docker Compose 本地起全栈（1 节点 PoC）

# docker-compose-vehicle-fullstack.yml
# 单节点 PoC 用：简化版，生产请用 K8s / Helm
version: "3.9"

services:
# — Kafka (KRaft mode, 3 节点简化为 1) —
kafka:
image: bitnami/kafka:latest
ports: ["9092:9092"]
environment:
KAFKA_CFG_NODE_ID: 1
KAFKA_CFG_PROCESS_ROLES: broker,controller
KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
ALLOW_PLAINTEXT_LISTENER: "yes"

# — MinIO（本地 S3，给 Paimon 用） —
minio:
image: minio/minio:latest
ports: ["9000:9000", "9001:9001"]
environment:
MINIO_ROOT_USER: minioadmin
MINIO_ROOT_PASSWORD: minioadmin
command: server /data ––console–address ":9001"

# — ClickHouse + Keeper（混部，单节点简化） —
clickhouse:
image: clickhouse/clickhouse–server:latest
ports: ["9000:9000", "8123:8123"]
volumes:
– ./ch_local.xml:/etc/clickhouse–server/config.d/local.xml
depends_on: [kafka, minio]
ulimits:
nofile: { soft: 262144, hard: 262144 }

# — Doris (FE + BE) —
doris-fe:
image: apache/doris:latest–fe
ports: ["8030:8030", "9030:9030"]
environment:
FE_SERVERS: "fe1:doris-fe:9010"
FE_QUORUM_VOTERS: "fe1:doris-fe:9010"
ADMIN_PASSWD: ""

doris-be:
image: apache/doris:latest–be
depends_on: [doris–fe]
environment:
FE_SERVER: "fe1:doris-fe:9010"

# — Superset (可选 BI) —
superset:
image: apache/superset:latest
ports: ["8088:8088"]
depends_on: [clickhouse, doris–fe]

启动命令：

# 起全栈
docker compose -f docker-compose-vehicle-fullstack.yml up -d

# 建 MinIO bucket（Paimon 用）
mc alias set myminio http://localhost:9000 minioadmin minioadmin
mc mb myminio/vehicle-lake

# 建 Kafka topic
docker compose exec kafka kafka-topics.sh –create \\
–bootstrap-server kafka:9092 \\
–topic vehicle_cleaned –partitions 8 –replication-factor 1

# 建 Doris BE（注册到 FE）
mysql -h localhost -P 9030 -u root -e "
ALTER SYSTEM ADD BACKEND 'doris-be:9050';
"

# 建 ClickHouse 表（第 16/17 章的 DDL 直接用）
clickhouse-client –host localhost –port 9000 < ch_ddl.sql

18.6.3 K8s / Helm 生产部署要点

生产环境用 Helm chart + StatefulSet。ClickHouse 官方有 ClickHouse Kubernetes Operator 和 ClickHouse Helm Chart。

ClickHouse Helm values.yaml 关键配置：

# values.yaml 节选
clickhouse:
replicas: 3 # 每 shard 3 副本
shards: 4 # 4 shard
keeper:
enabled: true
replicas: 3 # Keeper 3 节点混部
storage:
data:
storageClass: ssd # NVMe SSD storageClass
size: 4Ti
config:
remote_servers:
vehicle_cluster:
– shard:
weight: "1"
internal_replication: "true"
replicas:
– host: clickhouse–shard0–replica0
– host: clickhouse–shard0–replica1
– host: clickhouse–shard0–replica2
# shard 1, 2, 3 同结构
macros:
shard: "{{ .Values.shard }}"
replica: "{{ .Values.replica }}"
named_collections:
paimon_conf:
url: "s3://vehicle-lake/paimon-table/"
access_key_id: "xxx"
secret_access_key: "xxx"
ingress:
host: clickhouse.internal

Doris Helm 用 Apache Doris Official Helm Chart：

doris-fe:
replicas: 3
image: apache/doris:2.1.6
doris-be:
replicas: 3
storage:
storageClass: ssd
size: 4Ti
doris-cn: # 可选 Compaction Job
enabled: true

18.6.4 全链路验证 checklist

[ ] Kafka: topic vehicle_cleaned 创建成功，3 节点 replication-factor 生效
[ ] Keeper: 3 节点 quorum 全绿（mntr 返回 imok）
[ ] ClickHouse: 4 shard × 3 replica = 12 节点，clusterAllReplicas 查询全 is_active=1
[ ] ClickHouse: Kafka 表创建成功，kafk_num_consumers=4
[ ] ClickHouse: Materialized View 管道状态正常（system.macros 里能查到）
[ ] Paimon: Flink JobManager 启动成功，sink 指向 S3 vehicle-lake bucket
[ ] Doris: JDBC Catalog 创建成功，能 SELECT ClickHouse 的 vehicle_telemetry 表
[ ] 端到端: 发送 1000 条模拟遥测 → 检查 Kafka → 检查 Flink 处理 → 检查 ClickHouse vehicle_kafka_mv 写入 → 检查 Doris JDBC Catalog 能查到
[ ] 故障: 停掉 1 个 Keeper 节点 → ClickHouse 仍可写（quorum 剩 2）
[ ] 故障: 停掉 1 个 ClickHouse replica → 其他副本仍可写

18.7 车辆链路全链路实战：端到端跑通

18.7.1 场景与前置条件

我们把第 15~17 章的 ClickHouse 集群扩展到了 5 组件全栈：

T-Box → Kafka → Flink(清洗) → Paimon(S3) + ClickHouse(4 shard × 3 replica)
↓ ↓
PaimonS3 外表 Doris JDBC Catalog
↓ ↓
ClickHouse 本地 MV Superset BI 看板

前置：

• Kafka 3 节点（KRaft 模式）、MinIO（本地 S3）、ClickHouse 4 shard × 3 replica + Keeper 3 节点、Doris 3 FE + 3 BE 全部部署完成
• 第 16/17 章的 ReplicatedReplacingMergeTree 表 + Distributed 表 + 冷热分层 TTL 已配好
• Flink 1.17+（Flink Connector 兼容）+ Flink Kafka Connector + Flink Paimon Connector 已在 classpath 上

18.7.2 步骤 1：Flink ODS 清洗 Job

下面是一个完整的 Flink DataStream Job（车辆遥测 ODS 清洗 + 双 sink）：

// VehicleODSJob.java 完整骨架
public class VehicleODSJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE); // 60s checkpoint
env.getCheckpointConfig().setCheckpointStorage("s3://vehicle-lake/flink-checkpoints/");

// 1. Kafka Source（原始遥测）
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "kafka-1:9092,kafka-2:9092,kafka-3:9092");
kafkaProps.setProperty("group.id", "flink-ods-group");
kafkaProps.setProperty("auto.offset.reset", "latest");

DataStream<String> rawStream = env.fromSource(
KafkaSource.<String>builder()
.setBootstrapServers("kafka-1:9092")
.setTopics("vehicle_raw")
.setGroupId("flink-ods-group")
.setStartingOffsets(OffsetsInitializer.latest())
.setValueOnlyDeserializer(new SimpleStringSchema())
.build(),
WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5)),
"vehicle-kafka-source"
);

// 2. 清洗 + 转换
DataStream<VehicleTelemetryRecord> cleaned = rawStream
.map(new VehicleTelemetryParser()) // JSON → POJO + 字段补全
.filter(record ->
record.getSpeed() >= 0
&& record.getBatteryPct() >= 0 && record.getBatteryPct() <= 100
&& record.getReportedAt().isBefore(Instant.now().plusSeconds(60))
)
.name("clean-telemetry");

// 3. Sink 到 Paimon（历史湖仓）
// 需要 Flink Paimon Connector 0.8+
cleaned.sinkTo(
PaimonSink.forRowDataStream(
cleaned,
PaimonConfig.fromMap(Map.of(
"warehouse", "s3://vehicle-lake/",
"database", "demo",
"table", "vehicle_telemetry_paimon",
"write-mode", "upsert",
"primary-key", "vin,seq"
))
).build()
);

// 4. Sink 到 ClickHouse（实时聚合）
ClickHouseSink<VehicleTelemetryRecord> chSink = ClickHouseSink
.<VehicleTelemetryRecord>builder()
.withUrl("jdbc:clickhouse://ck-1-1:8123,ck-1-2:8123,ck-1-3:8123")
.withDatabase("demo")
.withTable("vehicle_telemetry")
.withUsername("flink_writer")
.withPassword("xxx")
.withMaxBatchSize(50000)
.withMaxInFlightRequests(3)
.withFlushInterval(Duration.ofSeconds(5))
.withShardingKeyExtractor(VehicleTelemetryRecord::getVin)
.withExactlyOnce(true)
.withConverter(VehicleTelemetryRecord::toClickHouseRow)
.build();
cleaned.sinkTo(chSink);

env.execute("Vehicle Telemetry ODS Job");
}
}

验收：

# 提交 Flink Job
flink run -d -c VehicleODSJob vehicle-ods-job.jar

# Flink Web UI (http://flink-jm:8081) 看 Source → Map → Filter → 两个 Sink 的 throughput

# Paimon 验证（MinIO bucket vehicle-lake/demo/vehicle_telemetry_paimon/ 有 snapshot 文件夹）
mc ls myminio/vehicle-lake/demo/vehicle_telemetry_paimon/snapshot/

# ClickHouse 验证
SELECT count() FROM demo.vehicle_telemetry;
— 预期：持续增长（每秒 ~10000 行）

18.7.3 步骤 2：Paimon 外表 + 本地 MV 管道

— 开实验性功能
SET allow_experimental_paimon_storage_engine = 1;

— 外表（只读，指向 Paimon snapshot）
CREATE TABLE IF NOT EXISTS demo.vehicle_telemetry_paimon
(
vin String, city_id UInt32, report_date Date,
reported_at DateTime, speed Float32, battery UInt32,
battery_pct UInt8, vehicle_status LowCardinality(String),
seq UInt64, is_deleted UInt8, ext String
) ENGINE = PaimonS3(
— 用 Named Collection
paimon_vehicle, — url + key + secret（在 config.xml 里配好）
'JSONEachRow'
);

— 本地归档表（ReplicatedMergeTree，按月分区做冷数据归档）
CREATE TABLE IF NOT EXISTS demo.vehicle_telemetry_archive
ENGINE = ReplicatedReplacingMergeTree(
'/clickhouse/tables/{shard}/demo/vehicle_telemetry_archive',
'{replica}', seq
)
PARTITION BY toYYYYMM(report_date)
ORDER BY (vin, report_date);

— MV 管道：外表 → 本地归档（每天凌晨触发全量拉取）
CREATE MATERIALIZED VIEW demo.vehicle_telemetry_archive_mv
TO demo.vehicle_telemetry_archive AS
SELECT * FROM demo.vehicle_telemetry_paimon;

验收：

— 外表直接查
SELECT count() FROM demo.vehicle_telemetry_paimon;

— 归档表查
SELECT
toYYYYMM(report_date) AS month,
count() AS rows,
formatReadableSize(sum(bytes_on_disk)) AS size
FROM system.parts
WHERE table = 'vehicle_telemetry_archive' AND active = 1
GROUP BY month
ORDER BY month;

18.7.4 步骤 3：Doris JDBC Catalog 联邦查询

— Doris FE 上执行
CREATE CATALOG vehicle_ck_catalog PROPERTIES (
"type" = "jdbc",
"jdbc_driver_url" = "https://github.com/ClickHouse/clickhouse-jdbc/releases/download/v0.6.0/clickhouse-jdbc-0.6.0-all.jar",
"jdbc_driver_class" = "com.clickhouse.jdbc.ClickHouseDriver",
"jdbc_url" = "jdbc:clickhouse://ck-1-1:8123,ck-1-2:8123,ck-1-3:8123/demo",
"jdbc_user" = "doris_reader",
"jdbc_password" = "xxx"
);

USE vehicle_ck_catalog;

— Doris BI 看板核心查询：每城市平均车速 + 电量趋势
SELECT
city_id,
report_date,
avg(speed) AS avg_speed,
avg(battery_pct) AS avg_battery,
count() AS sample_count
FROM demo.vehicle_telemetry
WHERE report_date >= today() – 7
GROUP BY city_id, report_date
ORDER BY report_date, city_id;

验收：Superset 面板能看到过去 7 天各城市车速趋势折线图。

18.7.5 步骤 4：端到端故障演练

18.8 工程要点

18.9 本章小结

本章把第 1~17 章的单产品视角扩展到了全链路五组件生产部署：

• 我们画出了完整的五组件架构图——Kafka 解耦缓冲 → Flink 流处理 → Paimon 湖仓原始数据 + ClickHouse 查询聚合 → Doris BI 看板。ClickHouse 在其中扮演枢纽角色（三条写入口 + 两个读出口）。

• 我们对比了 Flink → ClickHouse 三种写入方式：Flink Connector（生产首选、AsyncSink、Exactly-Once、10~100k maxBatchSize）、JDBC Sink（通用但慢 2~5 倍）、Kafka 引擎管道（解耦、At-Least-Once + ReplacingMergeTree = 等价 Exactly-Once）。推荐方式 A + C 组合互为灾备。

• 我们学会了用 ClickHouse Paimon/Iceberg/Hudi/Hive 外表读湖仓数据——全部只读、partition filter 下推、Named Collection 存凭证、外表 + 本地 MV 管道做热缓存。车辆场景选 Paimon（Flink 原生）。

• 我们掌握了 Doris ↔ ClickHouse 双向联邦：主流方向 A（Doris JDBC Catalog 读 ClickHouse 做 BI）+ 方向 B（ClickHouse JDBC 引擎读 Doris 小字典表做 JOIN）。方向 A 是 BI 看板标配、Doris Routine Load 高频同步到本地表才是高性能看板正解。

• 我们深入了 Kafka 引擎 + MV 管道 的本质——只订阅解析不存数据、At-Least-Once + ReplacingMergeTree = Exactly-Once 效果、kafka_skip_broken_messages 容错、两条写入路互为灾备。

• 我们给出了 Docker Compose PoC + K8s/Helm 生产 两套部署方案的完整配置（端口表、Compose yaml、Helm values）+ 10 项全链路验证 checklist + 5 项故障演练清单。

18.10 下章预告

第 19 章《全系列复盘：ClickHouse 的 19 章知识地图》 将是整个 19 章系列的最后一章——把第 1 至 18 章所有知识点做成一张知识地图（ASCII 全景图），按"基础引擎 → 查询优化 → Replication → Distributed → 全链路"五大模块分类，标注每章覆盖的核心知识点 + 它们之间的依赖关系。同时给出 10 道跨章节综合思考题，让你检验自己是不是真的串起来了。

18.11 官方参考资料

18.12 版本取证说明

本章全部 ClickHouse 侧内容（Flink Connector 参数、Kafka 引擎 SETTINGS、外表引擎命名规则、Named Collections、JDBC 引擎、Cloud 限制）均与官方文档逐字对照。Doris JDBC Catalog 内容基于 Doris 官方文档。下表按四档口径标注。

18.13 实操清单

思考题

本章完。下一章：第 19 章 全系列复盘：ClickHouse 的 19 章知识地图。