如何使用Flink中的单个窗口进行多个汇总？

Question

我是Flink的新手，我想做很多次在Spark中做的事情。

例如，在Spark中，我可以在

ds.groupByKey(???).mapGroups(???) // aggregate 1
  .groupByKey(???).mapGroups(???) // aggregate 2

第一个汇总交易之下进行类似的输入数据，而第二个汇总处理第一个汇总的输出。我需要的是第二个骨料的输出。

但是在flink中，似乎任何聚合物都应使用特定窗口执行，

ds.keyBy(???)
  .window(???) // window 1
  .aggregate(???) // aggregate 1
  .keyBy(???)
  .window(???) // window 2
  .aggregate(???) // aggregate 2

如我设置窗口2，那么第二个汇总的输入数据可能不是第一个汇总的输出，这将违反我的愿望。

我想使用相同的批处理数据进行多个连续汇总，可以在一个窗口中收集。如何在Flink中实现它？

感谢您的帮助。

---

更新以获取更多详细信息。

窗口必须具有自己的策略，例如，我可能会设定窗口策略，例如

ds.keyBy(key1)
  .window(TumblingProcessingTimeWindows.of(Time.of(1, TimeUnit.HOURS))) // window 1, 1 hour tumbling window
  .aggregate(???) // aggregate 1
  .keyBy(key2)
  .window(TumblingProcessingTimeWindows.of(Time.of(1, TimeUnit.MINUTES))) // window 2, 1 minute tumbling window
  .aggregate(???) // aggregate 2

1窗口下方的窗口策略，可能会在一个小时的时间窗口中收集10亿行，并且在汇总后，它会输出一百万行。

我想在总计2中对一百万行进行一些计算，但我不知道哪种窗口策略可以准确地收集一百万行。

如果我将窗口2设置为上面的翻滚时间窗口，则可能将一百万行分为两批，而汇总2的输出将不是我需要的。

Answer 1

您可以通过使用事件时间窗口而不是处理时间窗口来避​​免此问题。而且，如果您在要用作计时基础的事件中还没有时间戳，那么您可以执行此类操作以使用摄入时间时间戳：

WatermarkStrategy<MyType> watermarkStrategy =
        WatermarkStrategy
            .<MyType>forMonotonousTimestamps()
            .withTimestampAssigner(
                (event, streamRecordTimestamp) -> Instant.now());

DataStream<MyType> timestampedEvents = ds
        .assignTimestampsAndWatermarks(watermarkStrategy);

timestampedEvents.keyBy(...)
  .window(TumblingEventTimeWindows.of(Time.of(1, TimeUnit.MINUTES)))
  .aggregate(...)
  .keyBy(...)
  .window(TumblingEventTimeWindows.of(Time.of(1, TimeUnit.HOURS)))
  .aggregate(...)

这有效，因为第一窗口产生的事件将每个时间戳都用时间戳将其分配给窗口的末端。这就要求第二个窗口的持续时间与第一个窗口的持续时间相同或倍数。

同样，对窗口2使用的密钥分区进行任意更改（与窗口1相比）可能会产生荒谬的结果。