Flink流式事件时间窗口排序

Question

import java.util.concurrent.TimeUnit
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.scala._

object WindowExample extends App {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.getConfig.enableTimestamps()
    env.setParallelism(1)

    val start = 1449597577379L
    val tuples = (1 to 10).map(t => (start + t * 1000, t))

    env.fromCollection(tuples)
      .assignAscendingTimestamps(_._1)
      .timeWindowAll(Time.of(1, TimeUnit.SECONDS))
      .sum(1)
      .print()

    env.execute()
}

输入：

 (1449597578379,1)
 (1449597579379,2)
 (1449597580379,3)
 (1449597581379,4)
 (1449597582379,5)
 (1449597583379,6)
 (1449597584379,7)
 (1449597585379,8)
 (1449597586379,9)
 (1449597587379,10)

结果：

[info] (1449597579379,2)
[info] (1449597581379,4)
[info] (1449597583379,6)
[info] (1449597585379,8)
[info] (1449597587379,10)
[info] (1449597578379,1)
[info] (1449597580379,3)
[info] (1449597582379,5)
[info] (1449597584379,7)
[info] (1449597586379,9)

Answer 1

这种行为的原因是在Flink中没有考虑元素的顺序（相对于时间戳）。只有水印的正确性及其与元素时间戳的关系对于考虑时间的操作才是重要的，因为水印通常在基于时间的操作中触发计算。

在您的示例中，window运算符将源中的所有元素存储在内部窗口缓冲区中。然后，源发出一个水印，表示未来不会有时间戳较小的元素到达。这又告诉window操作符处理结束时间戳低于水印的所有窗口（这对所有窗口都是真的）。因此，它发出所有的窗口（任意排序），然后它发出一个水印本身。下游的操作将自己接收这些元素，并且一旦接收到水印就可以进行处理。

默认情况下，从源发出水印的间隔为200 ms。由于源发出的元素数量很少，所有元素都会在第一个水印发出之前发出。在一个实际的用例中，水印发射间隔比窗口大小小得多，您将得到窗口按照时间戳的顺序发射的预期行为。例如，如果每500毫秒有1小时窗口和水印。