Flink Interval Join状态清理行为相关技术问题咨询

你提供的示例SQL如下：

SELECT *
FROM Orders o, Shipments s
WHERE o.id = s.order_id
AND o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_time

问题1解答 #

• 答案是肯定的。如果o.order_time和s.ship_time只是普通时间列，没有声明为event time或processing time属性，Flink无法推导数据的过期规则，判断不出哪些历史数据不会再参与后续的Join匹配，因此会和普通Regular Inner Join一样永久留存两边表的所有历史数据到状态中。随着作业运行时间变长，状态体量会持续膨胀，很容易导致作业OOM、checkpoint耗时过长等稳定性问题，这种场景不建议直接用普通时间列做Interval Join。

问题2解答 #

• 答案是肯定的。如果两个时间列都已经声明为event time属性，且两张表都配置了合理的watermark生成策略，Flink完全可以依赖watermark自动清理过期状态，避免状态无限增长。
• 以上述SQL为例，Join的时间约束为o.order_time ≥ s.ship_time - 4小时且o.order_time ≤ s.ship_time：
  • 当Orders表的watermark超过某条订单数据的order_time + 4小时后，这条订单不可能再匹配到任何后续到来的运单数据，会被自动从状态中清除
  • 当Shipments表的watermark超过某条运单数据的ship_time后，这条运单不可能再匹配到任何后续到来的订单数据，也会被自动从状态中清除
• 只要watermark配置合理，状态体量始终可以维持在可控的较小范围。

内容的提问来源于stack exchange，提问作者Tom