ApacheFlink：窗口函数和时间的开始

Question

在WindowAsSigner中，元素被分配给一个或多个TimeWindow实例。在滑动事件时间窗口的情况下，这发生在SlidingEventTimeWindows#任务Windows¹中。

如果窗口的size=5和幻灯片=1，则将时间戳为0的元素分配到以下窗口：

1. 窗口（开始=0，结束=5）
2. 窗口（开始=-1，结束=4）
3. 窗口（开始=-2，结束=3）
4. 窗口（开始=-3，结束=2）
5. 窗口（开始=-4，结束=1）

在一幅图片中：

                            +-> Beginning of time
                            |
                            |
+----------------------------------------------+
|     size = 5              +--+ element       |
|    slide = 1              |                  |
|                           v                  |
| t=[ 0,5[ Window 1         XXXXX              |
| t=[-1,4[ Window 2        XXXXX               |
| t=[-2,3[ Window 3       XXXXX                |
| t=[-3,2[ Window 4      XXXXX                 |
| t=[-4,1[ Window 5     XXXXX                  |
|                                              |
| time(-4 to +4)        ----                   |
|                       432101234              |
+---------------------------+------------------+
                            |
                            |
                            |
                            +

有没有办法告诉Flink时间有开始，而在那之前，没有窗户？如果没有，从哪里开始寻求改变？在上述情况下，Flink第一个元素应该只有一个窗口（t=[4,8[Window 1）。像这样：

                            +-> Beginning of time
                            |
                            |
+-----------------------------------------------+
|     size = 5              +--+ element        |
|    slide = 1              |                   |
|                           v                   |
| t=[ 0,5[ Window 1         XXXXX               |
| t=[ 1,6[ Window 2          XXXXX              |
| t=[ 2,7[ Window 3           XXXXX             |
| t=[ 3,8[ Window 4            XXXXX            |
| t=[ 4,9[ Window 5             XXXXX           |
|                                               |
| time(-4 to +8)        ----                    |
|                       4321012345678           |
+---------------------------+-------------------+
                            |
                            |
                            |
                            +

一旦窗口数量达到并超过窗口大小，这将不再有效。然后，在上述情况下，所有元素都位于5个窗口内。

脚注：

1. org。阿帕奇。Flink。流动。应用程序编程接口。开窗。转让人。滑动事件时间窗口#分配窗口

Answer 1

我可能会为这个问题找到更好的解决方案。这个想法是把水印设置到将来足够远的地方，这样你的窗口就有足够的数据。早期的窗口仍然存在，但它们将被丢弃。

以下是具有周期性水印[T]的赋值器的概念证明：

  class WMG[T](wait: Long) extends AssignerWithPeriodicWatermarks[T] {
    var t: Option[Long] = None
    var firstTime = true

    override def extractTimestamp(el: T, prevTs: Long): Long = {
      t = Some(prevTs)
      prevTs
    }

    override def getCurrentWatermark(): Watermark = (t, firstTime) match {
      case (None, _) => return null
      case (Some(v), false) => new Watermark(v)
      case (Some(v), true) => {
        firstTime = false
        new Watermark(v + wait)
      }
    }
  }

更新：不幸的是，它不起作用（现在我不知道为什么应该），总是有几个键在键流与早期窗口。所以最后我只是过滤错误的窗口，比如：

val s = (winSize/winStep).intValue
kstream.flatMapWithState((in: StreamOut, state: Option[Int]) =>      
  state match {
    case None    => (Seq(), Some(1))
    case Some(s) => (Seq(in), Some(s))
    case Some(v) => (Seq(), Some(v+1))
  })

Answer 2

目前没有办法指定Flink作业的有效时间间隔。考虑到您可能也想将您的工作应用于历史数据，这也可能有点问题。

但是，您可以手动筛选在超时开始之前启动的窗口：

val env = StreamExecutionEnvironment.getExecutionEnvironment

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

val startTime = 1
val windowLength = 2
val slide = 1

val input = env.fromElements((1,1), (2,2), (3,3))
               .assignAscendingTimestamps(x => x._2)

val windowed = input
      .timeWindowAll(Time.milliseconds(windowLength), Time.milliseconds(slide))
      .apply{ (window, iterable, collector: Collector[Int]) =>
         if (window.getStart >= startTime) {
           collector.collect(iterable.map(_._1).reduce(_ + _))
         } else {
           // discard early windows
         }
       }

windowed.print()

env.execute()