双流 Join 乱序对齐

难度与前置

难度：架构
前置：读过 Watermark 与迟到、维表 Join；知道双流都是变更流时不能当维表查。
读完能做什么：用 Interval / 窗口 Join 对齐乱序双流，配置时钟与状态 TTL，并用延迟与匹配率验收。

适用场景

• 订单流 ⋈ 支付流、曝光 ⋈ 点击，两侧都是无限事件流
• 事件时间乱序，需要「在时间容差内」匹配
• 要明确迟到策略：侧输出 / 丢弃 / 修正

不适用：

• 一侧是缓慢变化维表（用 维表 Join）
• 纯批 Combine / 离线 Join
• 无事件时间、只按处理时间凑合（报表可，但对账难）

真实场景（一条）

订单与支付双流 Interval Join，容差 15min。支付网关时钟漂移 + Kafka 分区乱序，大量订单在边界外匹配失败，收入漏计。改造：统一事件时间字段（业务完成时间）；Watermark 允许 3min 乱序；Interval between -20min and 5min；迟到进侧输出日批补偿；状态 TTL 覆盖最大乱序+容差。匹配率从 96%→99.7%，与离线对账对齐。

怎么做（步骤清单）

1. 先分清 Join 形态

形态	适用	状态风险
Interval Join	键相等 + 时间区间	中：按键缓冲区间内事件
窗口 Join	共享窗口对齐	中
正则双流 Join + 时间约束	SQL Temporal / Interval	同左
维表 Lookup	一侧外部存储	低（见维表文）

2. 事件时间与 Watermark 对齐

orders.assignTimestampsAndWatermarks(
  WatermarkStrategy
    .<Order>forBoundedOutOfOrderness(Duration.ofMinutes(3))
    .withTimestampAssigner((e, ts) -> e.getEventTimeMs()));

payments.assignTimestampsAndWatermarks(
  WatermarkStrategy
    .<Pay>forBoundedOutOfOrderness(Duration.ofMinutes(3))
    .withTimestampAssigner((e, ts) -> e.getEventTimeMs()));

两侧 Watermark 推进节奏要匹配业务；一侧长期空闲会卡住 Join（空闲检测 withIdleness）

3. Interval Join 示例

orders.keyBy(Order::getOrderId)
  .intervalJoin(payments.keyBy(Pay::getOrderId))
  .between(Time.minutes(-20), Time.minutes(5))
  .process(new OrderPayMatchFn());

-- Flink SQL 示意
SELECT o.id, o.amt, p.pay_ts
FROM orders o, payments p
WHERE o.id = p.order_id
  AND p.pay_ts BETWEEN o.evt_ts - INTERVAL '20' MINUTE
                   AND o.evt_ts + INTERVAL '5' MINUTE;

4. 容差怎么定

容差 ≥ 业务最大合法延迟（支付可晚于下单）
     + 观测到的乱序 P99
     + 安全余量
过大 → 状态与延迟↑；过小 → 匹配率↓
用离线「真实匹配时差」直方图定参，勿拍脑袋

5. 迟到与补偿

• 主路径：Interval 内匹配输出
• 迟到：侧输出 → Kafka / 表，进 乱序与重复 补偿链路
• 日终：离线 Join 打底，实时做增量修正（与 对账 一致）

6. 状态与 TTL

• TTL ≥ 容差 + Watermark 乱序 + 停顿余量
• 键倾斜（头部 order_id 异常）会导致单键状态膨胀
• 监控 state size、Join 算子反压（反压）

7. 时钟与字段纪律

问题	处理
设备本地时钟	不用；用服务端事件时间
字段混用处理时间	禁止与事件时间 Join 混用
时区	统一 UTC 存，展示再转

常见坑（至少牢记这些）

1. 把双流当维表 Lookup
   只见当下不见历史支付，回放与乱序全错。

2. 两侧 Watermark 策略不一致
   一侧激进一侧保守，匹配窗口「看起来够」实际提前关闭。

3. 容差过小掩盖乱序
   业务说「支付晚 10 分钟正常」，你只给 2 分钟。

4. 无空闲检测
   某 key 或某源长时间无数据，Watermark 卡住，整作业延迟飙升（表象像 链路排查 里的水位问题）。

5. 状态 TTL 短于容差
   另一侧迟到到达时状态已清，静默丢匹配。

6. 忽略一对多
   一单多次支付未定义主支付选取规则，指标重复。

如何验收（可量化）

• [ ] 匹配率 ≥ 目标（如 99.5%），相对离线金标
• [ ] 端到端延迟 P99 ≤ SLA
• [ ] 迟到侧输出占比可解释，有补偿闭环
• [ ] 状态大小稳定，CK 成功
• [ ] 注入乱序/延迟事件：仍能在容差内匹配

-- 离线金标 vs 实时匹配
SELECT
  COUNT(*) AS offline_matched,
  SUM(CASE WHEN r.order_id IS NOT NULL THEN 1 ELSE 0 END) AS rt_hit
FROM (
  SELECT o.order_id
  FROM dwd_order_di o
  JOIN dwd_pay_di p ON o.order_id = p.order_id
  WHERE o.dt='${biz_date}'
    AND p.pay_ts BETWEEN o.evt_ts - INTERVAL 20 MINUTES AND o.evt_ts + INTERVAL 5 MINUTES
) g
LEFT JOIN dws_order_pay_rt_di r
  ON g.order_id = r.order_id AND r.dt='${biz_date}';

# 指标
echo "match_rate,late_sideoutput_rate,state_mb,p99_ms"
# 0.997,0.012,4096,800

Flink UI：
- Join 算子 watermark 两侧均推进
- 无长期 IDLE 卡住
- checkpoint 连续成功

相关面试题

占位：与题库小程序对齐后补链接（建议关联：双流 Join、Interval Join、Watermark、乱序对齐）。

---

改写说明：结构参考 Wood 原创实时实践，已改写成可执行清单；未包含未公开表名与内部阈值原值。

附：参数标定最小集

1. 从离线抽样 1～7 天，画 |pay_ts - order_ts| 分位数
2. 取 P99 作容差下界，再加 Watermark 乱序
3. 预发用同参回放一天，核对匹配率与状态体积后再上线