企业官网建设流程全解析

1. 双流Join的常规打法为什么让人头疼

做实时计算的同学，只要业务里出现过"两条流要按某个维度对齐"的需求，大概率都绕不开双流Join这个话题。Flink SQL里最直接的做法就是JOIN ... ON a.key = b.key，写起来跟离线SQL几乎一样，看起来人畜无害。但真正把它放到生产环境跑起来，你会发现事情远没有这么简单。

先说最典型的场景：一条是订单流，一条是支付流，业务上希望把同一个订单的支付信息补全到订单上。用Regular Join写，SQL层面确实能跑通，但状态会无限增长——因为Flink不知道两条流的数据什么时候算"到齐了"，只能把两边都缓存着等对方。时间一长，状态后端直接爆掉，Checkpoint越来越大，任务重启一次要恢复半天。这是很多人第一次踩双流Join的坑。

再往下走，你会接触到Interval Join，也就是给Join加一个时间窗口约束，比如WHERE a.ts BETWEEN b.ts - INTERVAL '5' MINUTE AND b.ts + INTERVAL '5' MINUTE。这个方案确实能控制状态规模，但它对时间语义的要求很严格，必须是事件时间，还得配好Watermark。一旦某条流出现数据延迟或者乱序，Watermark推进不动，窗口就迟迟不触发，下游迟迟拿不到结果。更麻烦的是，Interval Join只支持两条流，如果你有三条、四条流要按同一个维度关联，就得嵌套写，SQL复杂度直线上升。

还有一类场景是维表Join，比如订单流去关联一张MySQL的用户表。这个用Lookup Join能解决，但它是同步查询，QPS一高就会把维表库打挂，而且维表更新有延迟，实时性要求高的业务根本受不了。

所以问题就来了：有没有一种方式，既能控制状态规模，又能灵活处理多条流的关联，还不用被Watermark卡得死死的？这就是"用union all方式实现双流Join"这个思路的出发点。它本质上不是Flink SQL层面的Join算子，而是一种流合并+分组处理的工程化打法，把两条流先union成一条，再在同一个算子内按key做状态管理，自己控制关联逻辑和状态清理。听起来有点"返璞归真"，但实测下来在很多场景里比原生Join更稳、更可控。

这篇文章我会把这个方案的来龙去脉讲清楚：为什么union all能替代Join、状态怎么设计、代码怎么写、有哪些坑、什么场景适合用、什么场景千万别用。如果你正在被双流Join的状态问题折磨，或者你的Join条件比较复杂、原生SQL表达不出来，这篇应该能给你一条新路子。

2. union all实现双流Join的核心思路拆解

2.1 从"两条流"到"一条流"的思维转换

原生Join的思路是：两条流各自独立，Flink在算子层面帮你做匹配。而union all的思路是：我先把两条流合并成一条流，合并之后每条数据打上一个"来源标记"，然后在同一个处理逻辑里，根据key把数据分到不同的状态里，自己判断什么时候可以输出关联结果。

打个比方，原生Join像是两个人在两个房间里，中间有个传话筒帮你对暗号；union all像是把两个人拉到同一个房间，你自己拿着名单核对谁和谁是一对。前者依赖Flink的Join机制，后者把控制权完全交到你手里。

这个转换带来的第一个好处是：状态管理完全自主。你可以决定每条数据在状态里存多久、什么时候清理、清理策略是什么。原生Regular Join的状态清理是Flink内部控制的，你插不上手；而union all之后，状态是你自己维护的ValueState或MapState，想怎么清就怎么清。

第二个好处是：支持多流关联。三条流、四条流要按同一个key关联，原生Join得嵌套写，而union all只需要给每条流打上不同的来源标记，在同一个KeyedProcessFunction里统一处理就行。代码结构反而更清晰。

第三个好处是：Join条件更灵活。原生SQL的Join条件受限于SQL表达能力，比如你想做"模糊匹配"或者"一对多再聚合"的逻辑，SQL写起来很别扭。而union all之后，你拿到的是原始数据，想怎么匹配就怎么匹配，完全用Java/Scala代码控制。

2.2 来源标记与状态分区的设计

union all之后，最关键的设计是来源标记。每条数据从哪条流来的，必须带一个字段标识，比如sourceType，订单流打ORDER，支付流打PAY。这个标记决定了这条数据进哪个状态、触发什么逻辑。

状态设计上，通常用两个MapState或者ValueState，一个存订单侧的数据，一个存支付侧的数据，key就是关联字段（比如订单号）。当一条订单数据进来，先存到订单状态，然后去支付状态里查有没有对应的支付记录，有就输出关联结果；支付数据进来同理。

这里有个细节要注意：状态里存什么。如果只存关联字段，那关联上了也拿不到完整数据，所以状态里要存完整的业务对象，或者至少存输出结果需要的字段。但存太多字段又会导致状态膨胀，所以要在"存全"和"存少"之间做权衡。我的经验是，状态里存一个精简后的POJO，只保留输出必需的字段，其他字段在输出前从当前数据里取。

2.3 状态清理策略：TTL与主动清理的组合拳

union all方案最大的优势就是状态可控，但前提是你得真的去控制它。如果只是把数据往状态里塞，从不清理，那跟Regular Join没区别，照样爆。

状态清理一般两条路：TTL自动过期和主动清理。

TTL是Flink State自带的机制，配置StateTtlConfig，设置过期时间和刷新策略。比如订单数据在状态里最多存24小时，超过就自动删。这个适合"关联有时间窗口"的场景，比如订单和支付一般几分钟内就关联上了，超过1小时还没关联的基本就是异常单，可以清理。

主动清理是在代码里显式调用state.clear()或state.remove(key)。比如当一条订单和支付成功关联并输出后，如果业务上确认这对数据不会再被关联，就可以立即把状态清掉，释放空间。这个适合"一对一关联"的场景，关联完就完事。

实际生产中，我一般两个都用：TTL兜底，防止漏清；主动清理做优化，减少状态驻留时间。TTL设一个宽松的上限（比如24小时），主动清理在关联成功或超时后触发。

2.4 与原生Join的适用边界对比

不是所有场景都适合union all。我整理了一个对比表，帮你快速判断：

简单说：如果你的Join逻辑简单、数据量不大、状态能控制住，原生Join最省事；如果时间窗口明确、事件时间可靠，Interval Join很优雅；如果状态压力大、Join条件复杂、或者要多流关联，union all值得考虑。

3. 手把手实现：从数据建模到代码落地

3.1 数据模型与来源标记的定义

先定义两条流的数据结构。假设订单流和支付流都有关联字段orderId，订单流有orderId、userId、amount、orderTime，支付流有orderId、payId、payAmount、payTime。

union all之前，需要把两条流统一成同一个类型。有两种做法：一是定义一个宽泛的父类或统一POJO，包含所有字段，用不到的字段留空；二是用Either或自定义的包装类。我倾向于第一种，定义一个UnionEvent，包含sourceType、orderId、以及所有业务字段，简单直接。

public class UnionEvent { public String sourceType; // "ORDER" or "PAY" public String orderId; public String userId; public Double amount; public Long orderTime; public String payId; public Double payAmount; public Long payTime; // 构造函数、getter/setter省略 }

订单流的数据转换成UnionEvent时，sourceType设为ORDER，支付相关字段留null；支付流同理。这样两条流union之后，类型一致，可以直接进同一个处理算子。

3.2 两条流的union与keyBy操作

union的代码很直接：

DataStream<UnionEvent> orderStream = ...; DataStream<UnionEvent> payStream = ...; DataStream<UnionEvent> unionStream = orderStream.union(payStream); DataStream<UnionEvent> keyedStream = unionStream.keyBy(e -> e.orderId);

注意union要求两条流的类型完全一致，所以前面的统一转换是必须的。keyBy之后，同一个orderId的数据会落到同一个并行子任务，这是后续状态关联的基础。

这里有个容易忽略的点：union之后的数据流，Watermark怎么处理。如果两条流都有Watermark，union之后Flink会取两条流Watermark的最小值。如果你不依赖Watermark做窗口计算，这个不影响；但如果后续还有窗口操作，就要注意Watermark的推进可能被慢的那条流拖住。

3.3 KeyedProcessFunction中的状态关联逻辑

核心逻辑写在KeyedProcessFunction里。我一般用两个MapState，key是orderId，value是业务对象。为什么用MapState而不是ValueState？因为一个orderId可能对应多条支付记录（比如分次支付），MapState能存多条，ValueState只能存一条。

public class UnionJoinFunction extends KeyedProcessFunction<String, UnionEvent, ResultEvent> { private MapState<String, OrderInfo> orderState; private MapState<String, PayInfo> payState; @Override public void open(Configuration parameters) { MapStateDescriptor<String, OrderInfo> orderDesc = new MapStateDescriptor<>("orderState", String.class, OrderInfo.class); MapStateDescriptor<String, PayInfo> payDesc = new MapStateDescriptor<>("payState", String.class, PayInfo.class); // 配置TTL StateTtlConfig ttlConfig = StateTtlConfig .newBuilder(Time.hours(24)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build(); orderDesc.enableTimeToLive(ttlConfig); payDesc.enableTimeToLive(ttlConfig); orderState = getRuntimeContext().getMapState(orderDesc); payState = getRuntimeContext().getMapState(payDesc); } @Override public void processElement(UnionEvent event, Context ctx, Collector<ResultEvent> out) throws Exception { if ("ORDER".equals(event.sourceType)) { // 订单数据进来 OrderInfo order = convertToOrder(event); orderState.put(event.orderId, order); // 检查支付状态里有没有对应的支付 if (payState.contains(event.orderId)) { for (PayInfo pay : payState.get(event.orderId).values()) { out.collect(merge(order, pay)); } // 关联成功后清理支付状态（如果是一对一） payState.remove(event.orderId); } } else { // 支付数据进来 PayInfo pay = convertToPay(event); if (!payState.contains(event.orderId)) { payState.put(event.orderId, new HashMap<>()); } payState.get(event.orderId).put(pay.payId, pay); // 检查订单状态里有没有对应的订单 if (orderState.contains(event.orderId)) { OrderInfo order = orderState.get(event.orderId); out.collect(merge(order, pay)); // 关联成功后清理订单状态 orderState.remove(event.orderId); } } } }

这段代码有几个关键点。第一，open方法里配置TTL，24小时过期，OnCreateAndWrite表示每次写入刷新过期时间，NeverReturnExpired表示不返回已过期的数据。第二，订单数据进来时，先存自己，再查对方；支付数据进来同理。第三，关联成功后主动清理对方状态，减少状态驻留。

3.4 关联结果的输出与下游处理

merge方法负责把订单和支付拼成结果对象。这里要注意字段冲突的处理，比如订单和支付都有amount字段，输出时要么重命名，要么明确用哪个。我一般输出一个ResultEvent，包含orderId、userId、orderAmount、payAmount、payTime等字段，语义清晰。

输出之后，下游可以接Kafka、JDBC、或者继续做聚合。如果下游还要按orderId分组，可以再keyBy一次；如果只是落地，直接addSink就行。

有个细节：输出结果的时机。上面的逻辑是"后到的数据触发输出"，也就是订单先到、支付后到，支付到达时输出；支付先到、订单后到，订单到达时输出。这样保证只要两边都到齐，就一定会输出一次，且只输出一次（因为关联后清理了状态）。但如果业务上要求"订单到达时立即输出，支付信息后续补全"，那就得用不同的策略，比如订单到达时先输出一个占位结果，支付到达时再发一个更新事件。这个取决于业务对实时性和完整性的权衡。

4. 状态管理与性能调优的实战细节

4.1 状态后端选型对方案的影响

union all方案的状态完全由你控制，所以状态后端的选型直接影响性能和稳定性。Flink现在主流的状态后端是HashMapStateBackend（内存）和EmbeddedRocksDBStateBackend（RocksDB）。

如果状态量不大（比如几百万个key以内），用HashMapStateBackend，读写快，延迟低。但内存有限，状态一大就OOM。如果状态量很大（千万级、亿级key），必须用RocksDB，它把状态存在磁盘上，内存只做缓存，能撑住大状态，但读写有序列化开销，延迟比内存高。

我的经验是：预估状态规模再选。估算方法：key数量 × 每个key的状态大小 × 副本数。比如1000万个订单，每个订单状态1KB，那就是10GB，加上RocksDB的放大系数，实际磁盘占用可能20GB以上。这种量级必须RocksDB。如果只有几十万key，内存后端完全够用。

另外，RocksDB的writeBufferSize、maxWriteBufferNumber、blockCacheSize这些参数对性能影响很大，默认值往往偏保守，生产环境需要根据数据量调优。比如blockCacheSize设大一点，能减少磁盘读。

4.2 状态TTL配置的坑与调优

TTL配置看起来简单，但有几个坑。

第一个坑：TTL的清理不是实时的。Flink的TTL清理是惰性的，只有在访问状态时才会触发过期检查，或者依赖后台的清理线程。所以即使设了24小时TTL，状态也不会在24小时整点被清掉，可能延迟很久。如果对状态大小敏感，不能完全依赖TTL，得配合主动清理。

第二个坑：OnCreateAndWrite和OnReadAndWrite的区别。前者只在写入时刷新过期时间，后者读写都刷新。如果你的业务是"读多写少"，用OnReadAndWrite能让状态活得更久；如果是"写多读少"，用OnCreateAndWrite更省。选错了会导致状态过早过期或过期太慢。

第三个坑：TTL和Checkpoint的关系。TTL过期状态在Checkpoint恢复后可能"复活"，因为Checkpoint里存的是未过期的快照。如果对状态一致性要求高，要注意这个行为。

4.3 大key与数据倾斜的应对

keyBy之后，如果某个orderId的数据特别多（比如热门商品订单），就会形成大key，落到同一个并行子任务，导致该子任务负载过高，其他子任务空闲。这是典型的数据倾斜。

应对方法有几个。一是加盐打散：给key加一个随机前缀，比如orderId + "_" + random(0, N)，这样同一个orderId的数据会分散到N个子任务，但代价是关联逻辑变复杂——你得保证订单和支付用同样的盐值，否则关联不上。二是两阶段聚合：先局部关联，再全局关联，适合聚合类场景。三是调整并行度：如果倾斜不严重，提高并行度能缓解。

对于union all方案，我一般先用加盐打散，因为关联逻辑本身就在代码里，加盐只是改一下key的生成逻辑，相对好控制。但要注意，加盐后同一个orderId的订单和支付必须用相同的盐值，所以盐值不能随机生成，得根据orderId做哈希取模，保证一致性。

4.4 Checkpoint与状态恢复的注意事项

union all方案的状态是你自己维护的，所以Checkpoint的配置直接影响恢复能力。几个关键点：

Checkpoint间隔：太短会增加开销，太长会导致恢复时重放数据多。一般1到5分钟比较合适，具体看数据量和SLA。

Exactly-Once语义：如果下游是Kafka，开启Checkpoint就能保证Exactly-Once；如果下游是数据库，需要配合事务或幂等写入。

状态恢复：任务重启时，Flink会从最近的Checkpoint恢复状态。如果状态很大，恢复时间可能很长。RocksDB的增量Checkpoint能减少每次Checkpoint的数据量，加快恢复。

Savepoint：做版本升级或逻辑变更时，用Savepoint保存状态，升级后从Savepoint恢复。但要注意，如果你改了状态的数据结构（比如给POJO加字段），恢复时可能报序列化错误，需要做状态迁移或者兼容处理。

5. 那些年我踩过的坑与排查实录

5.1 状态不清理导致的内存暴涨

最早用这个方案时，我只配了TTL，没做主动清理，想着TTL能兜底。结果跑了一周，TaskManager内存从4G涨到12G，Checkpoint从几百MB涨到十几GB。排查发现，TTL虽然配了，但清理不及时，大量已关联的数据还留在状态里。

后来加了主动清理：关联成功后立即remove对方状态。内存立刻降下来，Checkpoint也稳定在2GB左右。这个教训是：TTL是兜底，不是主力。能主动清的一定要主动清，别指望TTL。

5.2 union之后Watermark不推进的诡异现象

有一次任务跑着跑着，下游窗口不触发了。排查发现，union之后Watermark一直不推进。原因是其中一条流（支付流）的数据源某个分区没有数据，Watermark一直停在初始值，union取最小值，导致整体Watermark卡住。

解决办法是给空闲分区配置withIdleness，让Flink在分区空闲一段时间后忽略它的Watermark。配置如下：

WatermarkStrategy.<UnionEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withIdleness(Duration.ofMinutes(1))

这个坑很隐蔽，因为两条流单独跑都正常，union之后才暴露。

5.3 关联结果重复输出的排查过程

有段时间下游反馈数据重复。排查发现，支付流有重发机制，同一条支付数据可能来两次，导致payState里存了两条相同payId的记录，订单到达时遍历输出，就输出了两次。

解决办法是在payState的value里用payId做去重，或者用MapState的key做去重（payState.get(orderId).put(payId, pay)，相同payId会覆盖）。另外，下游如果对幂等有要求，可以在输出时带上唯一ID，下游做去重。

5.4 状态序列化失败的版本升级事故

有一次升级任务，给OrderInfo加了一个字段，从Savepoint恢复时报序列化错误。原因是Flink的POJO序列化器对字段变更敏感，新增字段会导致反序列化失败。

解决办法有两个：一是用Avro或Protobuf做序列化，支持schema演进；二是升级时不改状态结构，新增字段从外部获取。我后来改用了Avro，虽然麻烦点，但版本兼容性好很多。

6. 什么场景该用、什么场景别用

6.1 适合union all方案的典型业务

多流关联：三条以上流按同一key关联，原生Join嵌套复杂，union all更清晰。

状态敏感：数据量大、状态容易爆，需要精细控制状态生命周期。

Join条件复杂：比如模糊匹配、一对多再聚合、条件分支，SQL表达困难。

时间语义不可靠：数据乱序严重、Watermark难推进，不想被时间窗口卡住。

结果实时性要求高：数据到达即触发，不等Watermark。

6.2 不建议使用的场景与替代方案

简单一对一关联、数据量小：直接用Regular Join，开发成本低。

时间窗口明确、事件时间可靠：Interval Join更优雅，代码量少。

维表关联：用Lookup Join或Async I/O，别用union all硬扛。

需要Exactly-Once且逻辑复杂：union all的状态管理复杂，容易出错，不如用Flink原生Join+TTL。

团队Flink经验不足：union all方案对状态管理、Checkpoint、序列化都有要求，新手容易踩坑，建议先用原生方案。

6.3 方案选型的决策清单

最后给一个决策清单，帮你快速判断：

1. 关联流数量是否超过2条？是→考虑union all。
2. 状态规模是否超过内存承受能力？是→考虑union all+RocksDB。
3. Join条件是否SQL难以表达？是→考虑union all。
4. 时间语义是否可靠？否→考虑union all。
5. 团队是否有状态调优经验？否→先用原生Join。
6. 业务是否要求数据到达即输出？是→考虑union all。

这个清单不是绝对的，实际选型还要结合数据量、SLA、运维成本综合判断。我的建议是：先用最简单的方案跑通，遇到瓶颈再优化。别一上来就上union all，除非你明确知道原生Join满足不了。

我个人在实际操作中的体会是，union all方案最大的价值不是性能，而是可控性。状态怎么存、怎么清、什么时候输出，全在你手里。这种可控性在复杂业务里比性能更重要，因为它意味着出了问题你能定位、能修复，而不是对着Flink内部机制干瞪眼。当然，代价是开发复杂度上去了，代码量多了，测试要更充分。所以用不用，取决于你对可控性的需求有多强。