Apache Flink:水印不随广播流进行

Question

有1个高通量Kafka流定义如下

val stream: DataStream[A] = flinkEnv
  .addSource(kafkaStreamSource)
  .assignTimestampsAndWatermarks(
     WatermarkStrategy
        .forBoundedOutOfOrderness[A](Duration.ofSeconds(0))
        .withIdleness(Duration.ofSeconds(5))
        .withTimestampAssigner(new SerializableTimestampAssigner[A] {
            override def extractTimestamp(element: A, recordTimestamp: Long): Long = {
                element.lastUpdatedAt
              }
            }
        )
  )
  .window(TumblingEventTimeWindows.of(Time.minutes(1)))
  .reduce(.......)

上述窗口操作符的水印正确转发。

上述窗口操作符中的DataStream[A]需要使用一些保存在某些S3文件中的信息来丰富。S3文件很少更新。

S3文件作为流读取，然后广播以丰富DataStream[a]中的元素。

 val textInputFormat = new TextInputFormat(new Path("s3 path...."))
 val enrichWithElements: BroadcastStream[EnrichWithElement] = flinkEnv.readFile(textInputFormat, "s3 path ...", FileProcessingMode.PROCESS_CONTINUOUSLY, 30)
    .map(s3Element => {
      EnrichWithElement(.....)
    })
    .broadcast(new MapStateDescriptor......)

然后连接这两个流，用EnrichWithElement类型的元素来丰富A类型的所有元素。

class EnrichedAProcess
    extends BroadcastProcessFunction[A,EnrichWithElement,EnrichedAElement] {
  override def processElement(
      value: A,
      ctx: Context,
      out: Collector[EnrichedAElement]): Unit = {
      .....
      out.collect(EnrichedAElement(....))
  }

  override def processBroadcastElement(
      value: EnrichWithElement,
      ctx: Context,
      out: Collector[EnrichedAElement]): Unit = {
      .........
  }
}
stream
  .connect(enrichWithElements)
  .process(new EnrichedAProcess)

EnrichedAProcess有2个输入。其中之一是不断转发水印，但广播流没有任何时间信息或水印。这导致EnrichedAProcess的水印根本无法转发，因为它的一个输入没有传入水印。

是否有办法指定EnrichedProcess的水印仅依赖于非广播输入。

Answer 1

操作员将多个输入信道设置其自己的水印，以从所有活动信道接收的最新水印的最小值。

您可以做的是将水印策略应用于始终返回MAX_WATERMARK作为其水印的广播流。（您不需要担心为该流分配时间戳。）