窗口操作如何与Flink中的key By一起工作？

Question

我在Kafka有这样的数据：

{
  "account": "iOS", //Possible values: iOS, android, web, windows
  "events": [
    {
      "timestamp": "2017-07-03T20:19:35Z"
    }
  ]
}

时间戳从2017-07-03T20:19:35Z到2017-07-03T20:22:30Z（大约3分钟）。我有一个Flink程序，它吸收了上面Kafka主题的数据：

object TestWindow {
  def main(args: Array[String]) = {
    val props = new Properties()
    props.put(...)

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    env.addSource(new FlinkKafkaConsumer010[String]("topic", new SimpleStringSchema(), props)).
      keyBy(jsonStr => {
        val jsonParser = new JsonParser()
        val jsonObject = jsonParser.parse(jsonStr).getAsJsonObject()
        jsonObject.get("account")
    }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
      .reduce({(v1, v2) =>
        println(v1 + " " + v2)
        ""
      })
    env.execute()
  }
}

我很难理解窗口化是如何与keyBy操作一起工作的。

我知道上面的keyBy操作将创建包含相同键元素的不同分区。但我不知道什么时候创建窗口，以及如何将每个分区添加到哪个窗口？

我的猜测是，当每个JSON对象到达时，它将由keyBy操作键入，然后根据该JSON对象的事件时间戳创建一个10秒的窗口？

因此，例如，如果一个时间戳为“2017-07-03T20:19:35Z”的JSON对象到达并拥有一个帐户类型的iOS，那么将为iOS创建一个键控分区，并为“2017-07-03T20:19:35Z”到“2017-07-03T20:19:45Z”创建一个窗口？还是每10秒创建一个窗口，而不考虑来自Kafka的JSON对象的事件时间戳？




Answer 1

您关于keyBy的假设是正确的keyBy根据定义的键属性对流进行分区，并根据每个键计算窗口。

在示例中使用的TumblingEventTimeWindow具有固定的窗口边框，即边框不依赖于数据的时间戳。10秒的滚动窗口将从[00:00:00.000, 00:00:10.000)，[00:00:10.000, 00:00:20.000)等创建窗口。到达窗口操作符的记录将被分配给与其时间戳相交的窗口。当窗口关闭时（操作员的本地时间通过窗口的结束时间戳），窗口的结果被计算并发出。请注意，窗口仅使用第一条记录实例化，即空窗口不会触发计算，也不能发出数据，例如零计数。

其他窗口类型，如会话窗口，具有数据驱动的边界。在会话窗口的情况下，所有彼此间隔不超过一段时间的记录都被分组到一个窗口中。