Flink SlidingEventTimeWindows不像预期的那样工作

Question

我将流执行配置为

object FlinkSlidingEventTimeExample extends App {
    case class Trx(timestamp:Long, id:String, trx:String, count:Int)
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    val watermarkS1 = WatermarkStrategy
      .forBoundedOutOfOrderness[Trx](Duration.ofSeconds(15))
      .withTimestampAssigner(new SerializableTimestampAssigner[Trx] {
          override def extractTimestamp(element: Trx, recordTimestamp: Long): Long = element.timestamp
      })
    val s1 = env.socketTextStream("localhost", 9999)
      .flatMap(l => l.split(" "))
      .map(l => Trx(timestamp = l.split(",")(0).toLong, id = l.split(",")(1), trx = l.split(",")(2), count = 1))
      .assignTimestampsAndWatermarks(watermarkS1)
      .keyBy(l => l.id)
      .window(SlidingEventTimeWindows.of(Time.seconds(20),Time.seconds(5))) // Not working
      //.window(SlidingProcessingTimeWindows.of(Time.seconds(20),Time.seconds(5))) // Working
      .sum("count")
      .print
    env.execute("FlinkSlidingEventTimeExample")
}

我已经定义了水印，但不知道为什么它没有产生任何东西。有人有什么想法吗？我的flink版本是1.14.0

我的build.sbt如下所示：

scalaVersion := "2.12.15"



libraryDependencies += "org.apache.flink" %% "flink-streaming-scala" % "1.14.0"
libraryDependencies += "org.apache.flink" %% "flink-runtime-web" % "1.14.0"
libraryDependencies += "org.apache.flink" %% "flink-clients" % "1.14.0"
libraryDependencies += "org.apache.flink" % "flink-queryable-state-runtime" % "1.14.0"

我从套接字输入数据(端口:9999)，如下所示：

1640375790000,1,trx1
1640375815000,1,trx2
1640375841000,1,trx3
1640375741000,1,trx4

试图提供比窗口大小更大的时间戳，但仍然无法工作。

Flink Web截图：网络用户界面 水印

Answer 1

先前的答案被删除；它是基于对设置错误的假设。

当事件时间窗口无法产生结果时，总是与水印有关。

输入中的时间戳对应于

December 24, 2021 19:56:30
December 24, 2021 19:56:55
December 24, 2021 19:57:21
December 24, 2021 19:55:41

所以有足够的数据触发几个滑动窗口的关闭。例如，trx2有一个足够大的时间戳，它可以生成一个足够大的水印，以关闭这些包含19:56:30的窗口：

19:56:15 - 19:56:34.999
19:56:20 - 19:56:39.999

但是，您的执行图如下所示：

​

​

问题是套接字源和接下来的任务(执行平面映射->映射->水印的任务)之间的再平衡。您的四个事件中的每一个都将转到水印策略的不同实例，而有些实例没有接收到任何事件。这就是为什么没有生成水印的原因。

相反，您要做的是将输入解析和水印生成链接到同一并行性的源上，从而使执行图看起来如下所示：

​

​

此代码将完成以下操作：

env
  .socketTextStream("localhost", 9999)
  .map(l => {
    val input = l.split(",")
    Trx(timestamp = input(0).toLong, id = input(1), trx = input(2), count = 1)
  })
  .setParallelism(1)
  .assignTimestampsAndWatermarks(watermarkS1)
  .setParallelism(1)
  .keyBy(l => l.id)
  .window(SlidingEventTimeWindows.of(Time.seconds(20), Time.seconds(5)))
  .sum("count")
  .print

通常，不需要在并行的情况下执行水印，但是水印生成器的每个实例都必须有足够的事件来处理，或者使用withIdleness进行配置。(如果每个实例都是空闲的，那么也不会得到任何结果。)