Apache Flink双流连接场景下Watermark与元素迟到判定疑问

问题1解答 #

不是。首先如果后续对streamC做事件时间窗口计算，没有给C配置时间戳分配器的前提下，Flink无法获取C元素的事件时间，作业会直接抛出异常无法运行，根本不会走到迟到判定的步骤。
如果后续用的是处理时间窗口，本身就不依赖事件时间和水印做迟到判定，和CombineAB的水印没有关联。
另外CombineAB的水印是A、B流的最小值仅代表该算子向下游传递的水印值，没有元素时间戳做对比的前提下，无法完成任何迟到判定逻辑。

问题2解答 #

不是。CombineAB算子本身的水印生成逻辑不会变，仍然取A、B两个流的水印最小值向下游传递。你在streamC后新增的assignTimestampsAndWatermarks是一个独立的下游算子，它会完全忽略上游传递过来的水印，按照你配置的C的时间戳字段和乱序容忍度生成全新的水印继续向下游传递，只是覆盖了上游的水印传递结果，没有改变CombineAB自身的水印逻辑。

问题3解答 #

存在，不需要额外在streamC后加时间戳和水印配置，直接在CombineAB输出元素时手动指定时间戳即可：
在processElement1和processElement2方法中，不要直接用out.collect输出，改用Context提供的带时间戳的输出方法：

// 以processElement1为例，processElement2同理
override def processElement1(elem: A, ctx:Context, out: Collector[C]) {
    val c = C(elem.x, elem.y, time.Now())
    // 手动给C元素设置事件时间戳，用你需要的updatedTime字段
    ctx.outputWithTimestamp(c, c.updatedTime)
}

这种方案下：

• C元素已经自带了事件时间戳，不需要额外配置时间戳分配器
• CombineAB的水印逻辑不变，仍然取A、B两流的水印最小值向下游传递
• 下游的迟到判定会自动用C自身的时间戳和CombineAB传递的水印对比计算，完全符合需求。

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