这是【实时数仓】系列第3篇。上篇讲CDC到Doris乱序覆盖，这篇讲Flink多表JOIN状态爆炸。

周日下午接到电话，出库包裹的实时任务告警了。

打开Flink Web UI一看：Checkpoint size 12GB，还在涨。RocksDB的write stall已经触发，处理延迟从秒级飙到分钟级。

这个任务关联了10张MySQL CDC源表——shipment_package、package_trajectory、warehouse、owner、carrier……最后写一张Doris宽表。跑了两个月，状态从几百MB涨到12GB。

一上来我先给所有表加了TTL 10天，结果第二天产品就投诉物流状态没更新。排查发现轨迹类数据15天才到，10天TTL早就把状态清了。

改成30天，轨迹数据保住了，状态开始疯涨。

TTL太小丢数据，太大爆内存。两头堵。

---

根因：Regular JOIN的状态是无界的

TTL两头堵的根因，是Regular JOIN的状态管理机制。

官方文档原文：“For streaming queries, the grammar of regular joins is the most flexible…However, this operation has important operational implications: it requires to keep both sides of the join input in Flink state forever.”

Regular JOIN的工作方式：

左表来一条 → 查右表状态 → 关联 → 输出
右表来一条 → 查左表状态 → 关联 → 输出
两边的数据都要永久保存在状态里

10表JOIN = 10个中间状态。而且是链式的——a JOIN b的结果再JOIN c，中间结果会放大。官方文档原文：“This often because the intermediate state created by a chain of joins is much larger than the input state itself.”

我的任务：10张表JOIN，状态从几百MB涨到12GB。TTL是补救措施，不是根本解法。

---

Flink 2.1的终极方案：MultiJoin

查Flink官方文档，发现2.1版本引入了一个新算子——MultiJoin。

官方文档原文：“In Flink 2.1, we introduce a new multi-join operator, an optimization designed to significantly reduce state size and improve performance for join pipelines that involve record amplification and large intermediate state. This new operator eliminates the need to store intermediate state for joins across multiple tables by processing joins across various input streams simultaneously. This ‘zero intermediate state’ approach primarily targets state reduction.”

核心原理：传统Regular JOIN是二元的（a JOIN b，结果再JOIN c），每一步都产生中间状态。MultiJoin是多元的——多张表同时JOIN，不存中间状态。

官方文档原文：“This technique exchanges a reduction in storage requirements for a corresponding increase in computational effort, as intermediate states are re-evaluated upon necessity.”

翻译：用计算换存储。中间状态不存了，需要的时候重新算。

官方10表JOIN基准测试数据：

对比项	传统Regular JOIN	MultiJoin
状态大小	有中间状态，polynomial增长	零中间状态，linear增长
性能	基准	2x ~ 100x+
Checkpoint	慢（状态大）	快（状态小）
恢复时间	慢	快（状态小，节点少）

适用场景（官方建议）：

• 多表JOIN共享至少一个共同join key
• 中间状态远大于输入源
• 共同join key的selectivity高（每个key对应的记录数少）
• 状态50GB+

不适用场景：

• 共同join key的selectivity低（大量记录共享同一个key）→ 重新计算开销大
• 仅支持streaming INNER/LEFT JOIN

启用方式：

-- 方式一：全局启用
SET 'table.optimizer.multi-join.enabled' = 'true';

-- 方式二：用hint指定特定表
SELECT /*+ MULTI_JOIN(a, b, c) */ *
FROM shipment_package a
JOIN package_trajectory b ON a.id = b.package_id
JOIN carrier c ON a.id = c.package_id
...

支持的join条件：

-- ✅ 支持：共享同一个key
A JOIN B ON A.key = B.key JOIN C ON A.key = C.key

-- ✅ 支持：通过传递性共享key
A JOIN B ON A.key = B.key JOIN C ON B.key = C.key

-- ❌ 不支持：没有共享key
A JOIN B ON A.key1 = B.key1 JOIN C ON B.key2 = C.key2

---

现实：你大概用不了

MultiJoin是Flink 2.1的特性。我们生产环境是Flink 1.17。

版本	MultiJoin	我的环境
Flink 1.17	❌	✅ 生产
Flink 1.19	❌	—
Flink 2.1	✅	—

升级Flink版本不是小事——Connector兼容性、状态迁移、SQL语法变更，至少要一个季度的测试周期。

而且MultiJoin目前还是experimental状态：“This is currently in an experimental state - optimizations and breaking changes might be implemented.”

所以MultiJoin是方向，但不是现在能用的方案。

---

现在能用的：分层治理

既然用不了MultiJoin，就只能分层治理——把10张表按JOIN方式分层，能省状态的先省。

第一层：Event Time Temporal JOIN（2张维度表）

warehouse和owner是维度表，变化频率低。用Regular JOIN会存所有历史版本，用Event Time Temporal JOIN只存since watermark的版本。

官方文档原文：“The versioned table will store all versions - identified by time - since the last watermark.”

-- 之前：Regular JOIN，两边永久保存状态
LEFT JOIN warehouse d
  ON a.tenant_id = cast(d.tenant_id as bigint) AND a.warehouse_id = d.id

-- 之后：Event Time Temporal JOIN，取watermark时刻的版本
LEFT JOIN warehouse FOR SYSTEM_TIME AS OF a.update_time d
  ON a.warehouse_id = d.id AND a.tenant_id = cast(d.tenant_id as bigint)

前提条件：

• 维度表是Versioned Table = PRIMARY KEY + WATERMARK + changelog source（CDC天然支持）✅
• 主表有Event Time属性（WATERMARK）✅
• join条件包含右表的PRIMARY KEY ✅

Event Time vs Processing Time：

	Event Time Temporal Join	Processing Time Temporal Join
Flink 1.19	✅ 支持	❌ 不支持（“not support yet”）
取什么版本	数据变更时刻的版本	Flink处理时刻的最新版本

官方文档原文（Processing Time）：“Currently, the FOR SYSTEM_TIME AS OF syntax used in temporal join with latest version of any view/table is not support yet.”

所以我们用Event Time——基于CDC的update_time，join的是"数据变更时刻"的维度版本。

第二层：Regular JOIN + STATE_TTL hint（8张事实表）

其他8张表是事实表或传递关联表，不能用Temporal JOIN，只能用Regular JOIN + TTL。

-- 用STATE_TTL hint为每张表单独设TTL
SELECT /*+ STATE_TTL('b'='30d', 'c'='30d', 'f'='30d', 'g'='30d', 'h'='30d', 'i'='30d', 'j'='30d') */ *
FROM shipment_package a
LEFT JOIN package_trajectory b ON b.package_id = a.id AND b.deleted = 0
LEFT JOIN carrier c ON c.package_id = a.id
LEFT JOIN data_dict f ON a.tenant_id = cast(f.tenant_id as bigint) AND d.code = f.dict_code
LEFT JOIN package_trajectory_detail g ON b.id = g.package_trajectory_id
LEFT JOIN shipment_package_statistics h ON a.id = h.package_id
LEFT JOIN abnormal_package i ON a.id = i.package_id
LEFT JOIN shipment_package_consignee j ON a.id = j.package_id

官方文档原文：“For stateful computation Regular Join and Group Aggregation, users can use STATE_TTL hint to specify operator-level Idle State Retention Time, which enables the aforementioned operators to have a different TTL against the pipeline level configuration table.exec.state.ttl.”

还能做什么：MiniBatch Regular Join

官方文档原文：“By default, regular join operator processes input records one by one…This processing pattern may increase the overhead of StateBackend. Besides, this can lead to severe record amplification, especially in cascading join scenarios.”

MiniBatch Regular Join把输入缓存起来批量处理，减少record amplification：

SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.allow-latency' = '5S';
SET 'table.exec.mini-batch.size' = '5000';

---

优化效果

对比项	优化前	优化后
维度表JOIN	Regular JOIN（无界状态）	Event Time Temporal JOIN（watermark控制）
事实表JOIN	全局TTL（一刀切）	STATE_TTL hint（每张表单独设）
JOIN执行	逐条处理	MiniBatch批量处理

但这只是治标。 维度表的状态由watermark控制了，但8张事实表的状态还是靠TTL。TTL的两头堵问题依然存在——只是范围缩小了。

---

写在最后

Flink多表JOIN状态爆炸，不是"加TTL"就能解决的。

三层解法：

层	方案	适用版本	效果
终极	MultiJoin（零中间状态）	Flink 2.1+	2x~100x+性能，状态linear增长
过渡	Temporal JOIN + STATE_TTL + MiniBatch	Flink 1.17+	维度表状态有界，事实表靠TTL
临时	全局TTL	任何版本	两头堵

我们大部分人在第二层甚至第三层。MultiJoin是方向，但升级版本不是一朝一夕的事。

在此之前，能做的就是分层治理——维度表用Temporal JOIN省状态，事实表用STATE_TTL hint精细化TTL，开启MiniBatch减少record amplification。

写这篇之前，我也幻想过有一个配置项能把状态直接压下去。查了一圈Flink文档，发现2.1确实有——MultiJoin，零中间状态，10表JOIN性能提升2到100倍。但我们是1.17，这个配置项对我们来说就是一行不存在的代码。

所以这篇文章的结论其实很朴素：用不了终极方案的时候，把能省的先省了，把能分的先分了。维度表用Temporal JOIN，事实表精细化TTL，开启MiniBatch。不完美，但比全靠TTL补救强。

状态管理这件事，没有银弹，只有分治。

---

官方文档

• Joins: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/reference/queries/joins/
• Multiple Regular Joins / MultiJoin: https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/tuning/#multiple-regular-joins
• STATE_TTL Hints: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/reference/queries/hints/#state-ttl-hints
• Event Time Temporal Join: https://nightlies.apache.org/flink/flink-docs-master/docs/sql/reference/queries/joins/#event-time-temporal-join

下一篇讲Flink实时ETL——CDC整库同步、Schema Evolution、状态管理。