Flink 2.3.0 从理论到实践 —— 第 12 章 物化表与 SQL 高级特性

Flink 2.3.0 从理论到实践 —— 第 12 章 物化表与 SQL 高级特性

课程定位：连接器把数据接进来后，如何用 SQL 高级能力加速聚合与查询。本章讲解物化表（Materialized Table）概念与刷新策略、Flink 2.3 新特性 START_MODE 与 ALTER MATERIALIZED TABLE、连续查询与 Retract/Upsert 流、Mini-Batch 聚合与局部聚合、执行计划查看与查询优化、SQL 性能调优（并行度/内存/状态）——这是实时数仓"秒级出数 + 状态不爆"的调优总纲。

版本基线：Flink 2.3.0 + Paimon 1.4.x + Doris 4.1

章节导读

• 12.1 物化表（Materialized Table）概念
• 12.2 CREATE MATERIALIZED TABLE 与刷新策略
• 12.3 Flink 2.3 新特性：START_MODE 与 ALTER MATERIALIZED TABLE
• 12.4 连续查询与 Retract / Upsert 流
• 12.5 Mini-Batch 聚合与局部聚合
• 12.6 查询优化：执行计划查看与优化
• 12.7 SQL 性能调优：并行度、内存、状态
• 12.8 本章小结与下章预告

12.1 物化表（Materialized Table）概念

12.1.1 什么是物化表

物化表（Materialized Table，FLIP-550）是 Flink 2.x 引入的预计算结果集：把一个 SQL 查询的结果物化到一张持久化表，按策略自动刷新，查询走物化副本而非重算。

┌──────────────────────────────────────────────────────────────┐
│ 物化表 vs 普通表 vs 视图 │
└──────────────────────────────────────────────────────────────┘

普通表(Table) 视图(View) 物化表(Materialized Table)
存原始数据 只存 SQL 定义 存"查询结果"的快照
查询现算 查询展开重算 查询走物化副本
无刷新 无刷新 按策略自动刷新

12.1.2 物化表 vs Doris 物化视图

重要区分：Flink 物化表（Materialized Table）与 Doris 物化视图（Materialized View）是两套东西，别混淆。

项目实践：车联网项目里，Flink 物化表用于实时数仓分层（DWD→DWS 的预聚合结果物化，下游作业直接读）；Doris 物化视图用于 BI 加速——在离线 Paimon 表之上建异步 MV（BUILD DEFERRED，每日凌晨分区级 REFRESH），查询走 MV 本地副本毫秒级。两者各管一段，不冲突。

12.2 CREATE MATERIALIZED TABLE 与刷新策略

12.2.1 基本语法

CREATE MATERIALIZED TABLE m_dws_trip_daily — 物化表名
PARTITIONED BY (dt) — 分区列
REFRESH MODE — 刷新模式
FRESHNESS = INTERVAL '1' HOUR — 新鲜度(数据延迟上限)
AS SELECT
vin, dt, COUNT(*) AS trip_cnt, SUM(distance) AS total_dist
FROM paimon.tsp_ods.ods_vehicle_qb_msg_report_original
GROUP BY vin, dt;

12.2.2 三种刷新模式

┌──────────────────────────────────────────────────────────────┐
│ 物化表刷新模式 │
└──────────────────────────────────────────────────────────────┘

① CONTINUOUS(连续) ② MANUAL(手动) ③ FULL(全量)
流式常驻刷新 定时/手动触发 整表重算
FRESHNESS 承诺 调度器触发 每次全量

选型：实时需求 → CONTINUOUS；离线日级 → MANUAL（配 DolphinScheduler 调度）；种子校准 → FULL。

12.3 Flink 2.3 新特性：START_MODE 与 ALTER MATERIALIZED TABLE

12.3.1 START_MODE：刷新起点控制

Flink 2.3（FLIP-550/557）新增 START_MODE，控制物化表首次刷新从哪里开始读：

CREATE MATERIALIZED TABLE m_dws_trip_daily
PARTITIONED BY (dt)
REFRESH MODE = MANUAL
FRESHNESS = INTERVAL '1' HOUR
START_MODE = INITIAL — ★ 2.3 新增:首次全量
AS SELECT ... FROM paimon.tsp_ods.ods_msg;

项目硬约束类比：这与 Paimon 源表的 scan.mode 同源——实时作业读 Paimon 源必须 scan.mode='latest'（非默认 latest-full），否则全量回放 OOM。物化表的 START_MODE=INITIAL 仅用于首次初始化，常驻实时作业用 LATEST。

12.3.2 ALTER MATERIALIZED TABLE：DDL 演进

Flink 2.3（FLIP-557）支持 ALTER MATERIALIZED TABLE，可在线修改物化表定义而不丢失已物化的结果：

— 增加一列(物化结果保留,新列增量填充)
ALTER MATERIALIZED TABLE m_dws_trip_daily ADD COLUMN avg_speed DOUBLE;

— 修改刷新新鲜度
ALTER MATERIALIZED TABLE m_dws_trip_daily SET FRESHNESS = INTERVAL '30' MINUTE;

— 切换刷新模式
ALTER MATERIALIZED TABLE m_dws_trip_daily SET REFRESH MODE = CONTINUOUS;

2.3 价值：之前改物化表要 DROP + CREATE + 重算全量，2.3 支持增量演进——加列只回填新列，旧列保留，大幅降低变更成本。

12.3.3 与 Doris MV 的刷新对比

项目硬约束（反面教训）：Doris MV 刷新非分区表不要改分区表——hist/ads_stat 这类"当前值快照表"（PK=vin、整表覆盖、vin 点查）加 dt 分区是负优化：主键被迫变成 (vin,dt) 语义漂移、行数×保留天数、Doris MV 全量刷新成本×天数。判定线：能配 TTL 过期的表才适合按天分区。

12.4 连续查询与 Retract / Upsert 流

12.4.1 动态表与连续查询

Flink SQL 的核心抽象是动态表（Dynamic Table）：随时间不断变化的表。一个 SQL 查询在动态表上运行就是连续查询（Continuous Query）——结果表也随源表变化持续更新。

┌──────────────────────────────────────────────────────────────┐
│ 动态表 + 连续查询模型 │
└──────────────────────────────────────────────────────────────┘

源动态表(Kafka) 连续查询 结果动态表
┌──────────────┐ ┌──────────┐ ┌──────────────┐
│ vin=A cnt=1 │ ──────► │ GROUP BY │ ─────► │ vin=A cnt=2 │
│ vin=A cnt=2 │ 持续 │ vin │ 更新 │ vin=B cnt=1 │
│ vin=B cnt=1 │ ──────► │ │ ─────► │ │
└──────────────┘ └──────────┘ └──────────────┘

12.4.2 Changelog 四种操作

结果动态表的更新用 changelog（变更日志）表达，四种操作：

12.4.3 Append 流 vs Retract 流 vs Upsert 流

— 查询 1:Append 流(只追加,无更新)
SELECT vin, event_time FROM kafka_msg WHERE speed > 100;

— 查询 2:Retract 流(GROUP BY 产生更新/删除)
SELECT vin, COUNT(*) AS cnt FROM kafka_msg GROUP BY vin;
— 输出: +I(vin=A,1) → -U(vin=A,1) +U(vin=A,2) → …

— 查询 3:Upsert 流(GROUP BY 主键,无删除)
SELECT vin, MAX(speed) AS max_speed FROM kafka_msg GROUP BY vin;
— 主键 vin,只 +I/+U,无 -U/-D

12.4.4 流类型与 Sink 匹配

项目硬约束：车联网实时作业写 Doris Unique Key 表用 Upsert 流（+I/+U），sink.properties.partial_columns=true 部分列更新；写 Paimon 主键表用 Retract 流配 changelog-producer=full-compaction，可见性 ≤ Checkpoint 间隔（30s）。Retract 流写普通 Kafka topic 用 2.3 的 TO_CHANGELOG（第 11 章）。

12.5 Mini-Batch 聚合与局部聚合

12.5.1 Mini-Batch：高频更新降批量

Mini-Batch 把毫秒级单条更新攒成秒级批量更新，大幅降低下游写压力与状态访问频率。

┌──────────────────────────────────────────────────────────────┐
│ Mini-Batch 攒批原理 │
└──────────────────────────────────────────────────────────────┘

无 Mini-Batch: 有 Mini-Batch(size=5000 / latency=5s):
t=00:01 +U(vin=A,1) ┌── 攒 5s 或 5000 条 ──┐
t=00:02 +U(vin=A,2) │ │
t=00:03 +U(vin=A,3) ▼ │
t=00:04 +U(vin=A,4) t=00:05 一次性 +U(vin=A,5) │
t=00:05 +U(vin=A,5) (状态访问 1 次,下游写 1 次) │
(状态访问 5 次,下游写 5 次) │

项目硬约束：车联网 9 个常驻流作业统一开 Mini-Batch（size 5000 / latency 5s）。效果：状态访问从"每条一次"降到"每 5s 一次"，反压明显缓解，下游 Doris/Paimon 写入压力骤降。这是项目性能优化清单里"立竿见影"的一项。

12.5.2 Local-Global 两阶段聚合

Mini-Batch 之上还有 Local-Global（两阶段聚合）：先在各 SubTask 本地预聚合（Local），再按 key shuffle 后全局聚合（Global），减少网络 shuffle 数据量。

— 开启 Local-Global(默认开启,确认)
SET 'table.optimizer.agg-phase.enamble' = 'two_phase'; — 两阶段
— 或单阶段(关闭,倾斜场景慎用)
SET 'table.optimizer.agg-phase.enamble' = 'one_phase';

┌──────────────────────────────────────────────────────────────┐
│ Local-Global 两阶段聚合 │
└──────────────────────────────────────────────────────────────┘

源数据(分散) Local 预聚合 Shuffle Global 聚合
┌──────────┐ ┌──────────┐ ┌──────────┐
│ A:1 B:2 │─Local──►│ A:3 B:2 │──keyBy──►│ A:final │
│ A:2 C:1 │─Local──►│ C:1 │──keyBy──►│ B:final │
│ B:1 A:1 │─Local──►│ A:1 B:1 │──keyBy──►│ C:final │
└──────────┘ └──────────┘ └──────────┘
(N 条原始) (M 条预聚合,M<<N) (分组数条)

与 Mini-Batch 协同：Mini-Batch 攒批 → Local 本地预聚合 → shuffle → Global 全局聚合。三者叠加，高频更新作业的状态与网络都可控。

12.5.3 倾斜场景：两阶段聚合 + 加盐

数据倾斜时（某 key 数据量远超其他），Local-Global 不够用，需配合加盐：

— 倾斜 key 加盐打散,聚合后去盐还原
— 第一步:加盐 key + 局部聚合
SELECT
CONCAT(vin, '_', CAST(FLOOR(RAND() * 10) AS STRING)) AS salted_vin,
COUNT(*) AS partial_cnt
FROM ods_msg GROUP BY CONCAT(vin, '_', CAST(FLOOR(RAND() * 10) AS STRING));

— 第二步:去盐 + 全局聚合
SELECT
SUBSTR(salted_vin, 1, 17) AS vin, — 去 _N 后缀
SUM(partial_cnt) AS total_cnt
FROM step1 GROUP BY SUBSTR(salted_vin, 1, 17);

12.6 查询优化：执行计划查看与优化

12.6.1 EXPLAIN：查看执行计划

— 查看逻辑 + 物理执行计划
EXPLAIN
SELECT vin, COUNT(*) AS cnt
FROM ods_msg
WHERE dt = '2026-10-07'
GROUP BY vin;

— 查看优化后的执行计划(含 Optimizer 变换)
EXPLAIN PLAN;

输出结构（简化）：

== Optimized Physical Plan ==
TableSink(…) — Sink
└─ GroupAggregate(…) — 全局聚合
└─ Exchange(…)
└─ LocalGroupAggregate(…) — 本地预聚合
└─ Calc(…)
└─ Source(…)

12.6.2 Flink 优化器四层

┌──────────────────────────────────────────────────────────────┐
│ Flink SQL 优化器四层 │
└──────────────────────────────────────────────────────────────┘

① Parser(Calcite) SQL 文本 → SqlNode(未优化 AST)
② Validator SqlNode → RelNode(校验类型/表存在)
③ Optimizer RelNode → 优化后 RelNode
├─ 基于规则(RBO) 谓词下推/列裁剪/常量折叠
└─ 基于代价(CBO) Join 重排/聚合方式选择
④ Planner RelNode → ExecNode(物理计划)

12.6.3 常见低效写法与优化

项目硬约束（CURRENT_DATE 坑）：Flink 的 CURRENT_DATE 返回 DATE 类型，DATE_FORMAT 只接受 TIMESTAMP/CHARACTER，Calcite 直接拒绝：

SqlValidatorException: Cannot apply 'DATE_FORMAT' to arguments of type 'DATE_FORMAT(<DATE>, …)'

解法：CAST(CURRENT_DATE AS STRING)。注意 Doris 里 DATE_FORMAT(CURRENT_DATE(),…) 合法——两引擎别混写。

项目硬约束（改了 SQL 行为不变）：曾遇"改了 SQL 但行为不变"——根因是资源中心目录树与调度登记表【域】不一致，改的文件根本没被读到。提交脚本要打印 SQL 指纹（如行数 + 哈希），比对后再排查逻辑；本地仓库是唯一真相源。

12.7 SQL 性能调优：并行度、内存、状态

12.7.1 并行度四优先级

┌──────────────────────────────────────────────────────────────┐
│ 并行度优先级(高 → 低) │
└──────────────────────────────────────────────────────────────┘

① SQL Hint(表级) /*+ OPTIONS('parallelism'='6') */
② 算子级配置 table.exec.default-parallelism
③ 作业级配置 -D pipeline.parallelism (提交参数)
④ 全局默认 flink-conf.yaml 的 parallelism.default

项目硬约束：Source 并行度 = 上游 Kafka 分区数，Sink 并行度 = 下游 Paimon bucket 数（对齐避免数据倾斜）。如车联网 ODS bucket=6，则 Sink 算子并行度=6。表级 Hint 优先于作业级。

12.7.2 状态调优

项目硬约束：state.ttl=48h 平衡"断天归零"与状态膨胀——48h 内可从 checkpoint 恢复，超过则状态清理。配 HashMap backend（非 RocksDB），状态不无限膨胀。递推累计表用 LOOKUP JOIN _cum_rt 替代全历史状态，状态规模从"全历史"降到"当日"。

12.7.3 Checkpoint 调优

项目硬约束：Checkpoint 30s 与 Paimon 的 checkpoint commit 对齐——Paimon sink 靠 checkpoint 提交，可见性 ≤ 30s（changelog-producer=full-compaction）。间隔过短→反压；过长→故障丢数据多。30s 是车联网项目实测平衡点。

12.7.4 内存调优

TaskManager 内存模型：

┌──────────────────────────────────────────────────────────────┐
│ TaskManager 内存模型 │
└──────────────────────────────────────────────────────────────┘

Total Process Memory
├─ Framework Memory(框架) 固定
├─ Task Heap(算子堆) ★ SQL 算子 + 状态(HashMap)
├─ Task Off-Heap(直接内存) 网络缓冲 + RocksDB
├─ Managed Memory(托管) ★ Sort / RocksDB / 缓存
├─ Network Memory(网络) 输入输出缓冲
└─ JVM Metaspace / Overhead

12.7.5 性能优化清单（项目实测汇总）

12.8 本章小结与下章预告

本章小结

┌────────────────────────────────────────────────────────────────┐
│ 第 12 章 要点回顾 │
└────────────────────────────────────────────────────────────────┘

✓ 物化表(Materialized Table):
查询结果的持久化快照,按策略刷新
≠ Doris 物化视图(两套东西,别混)

✓ CREATE MATERIALIZED TABLE:
REFRESH MODE: CONTINUOUS(流) / MANUAL(调度) / FULL(全量)
FRESHNESS: 新鲜度承诺

✓ Flink 2.3 新特性:
START_MODE: INITIAL(首次全量) / LATEST(续接) /
FROM_TIMESTAMP / FROM_SNAPSHOT
ALTER MATERIALIZED TABLE: 加列增量回填,不丢旧结果
★ 非分区表不要改分区表(判定:能配 TTL 才按天分区)

✓ 连续查询与 changelog:
四操作: +I / -U / +U / -D
三流: Append / Retract / Upsert
Sink 匹配: Doris Unique Key 收 Upsert
Paimon 主键表收 Retract(changelog-producer)
普通 Kafka 用 TO_CHANGELOG(2.3)

✓ Mini-Batch + Local-Global:
mini-batch: size 5000 / latency 5s(项目实测)
Local 预聚合 → shuffle → Global 聚合
倾斜: 加盐两阶段聚合

✓ 查询优化:
EXPLAIN 看执行计划
优化器四层: Parser → Validator → Optimizer(RBO+CBO) → Planner
低效写法: SELECT * / 函数阻止分区裁剪 / Regular Join 全量状态
★ CURRENT_DATE 是 DATE,DATE_FORMAT 不收 → CAST AS STRING

✓ 性能调优:
并行度四优先级: Hint > 算子 > 作业 > 全局
Source 并行 = Kafka 分区数; Sink 并行 = bucket 数
state.ttl 48h + HashMap backend
checkpoint 30s(Paimon 可见性对齐)
★ 改 SQL 行为不变 → 提交脚本打印指纹,本地仓库是唯一真相源

下章预告

第 13 章 Kafka 实时读写：物化表与调优铺完后，进入第五篇数据集成。本章深入 Kafka Connector 全套——Source 消费位点与消费组、分区发现、Sink 幂等写入与 Exactly-Once 事务、JSON/Avro/Canal-JSON 格式、Kafka 与 Watermark 集成、背压与限速、Flink 2.3 Kafka Connector 新特性。

官方参考资料

• Materialized Table：https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/materialized-table/
• CREATE MATERIALIZED TABLE：https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/create/#create-materialized-table
• Dynamic Table & Changelog：https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/streaming/dynamic-tables/
• Query Optimization：https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/tuning/
• EXPLAIN：https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sql/explain/
• Configuration：https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/config/