title: How does the window operation work with the keyBy in Flink?
body: <p>I have data in Kafka that looks like this:</p>
  
  <pre><code>{
    "account": "iOS", //Possible values: iOS, android, web, windows
    "events": [
      {
        "timestamp": "2017-07-03T20:19:35Z"
      }
    ]
  }
  </code></pre>
  
  <p>The timestamp goes from <code>2017-07-03T20:19:35Z</code> to <code>2017-07-03T20:22:30Z</code> (about 3 minutes). I have this Flink program that ingests the data from the kafka topic above:</p>
  
  <pre><code>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 =&gt; {
          val jsonParser = new JsonParser()
          val jsonObject = jsonParser.parse(jsonStr).getAsJsonObject()
          jsonObject.get("account")
      }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
        .reduce({(v1, v2) =&gt;
          println(v1 + " " + v2)
          ""
        })
      env.execute()
    }
  }
  </code></pre>
  
  <p>I am having trouble understanding how windowing works with the <code>keyBy</code> operation.</p>
  
  <p>I understand that the <code>keyBy</code> operation above will create different partitions containing elements of the same key. But I am confused on when the windows are created and how each partition is added to which window? </p>
  
  <p>My guess is that as each JSON object arrives it will be keyed by the <code>keyBy</code> operation and then based on the event <code>timeStamp</code> of that JSON object it will create a window for 10 seconds?</p>
  
  <p>So for example if a JSON object with a timestamp of <code>"2017-07-03T20:19:35Z"</code> arrives and has a <code>account</code> type of iOS then a keyed by partition for iOS will be created and a window will also be created for <code>"2017-07-03T20:19:35Z"</code> to <code>"2017-07-03T20:19:45Z"</code>? Or is a window created every 10 seconds regardless of the event timestamp that is from the JSON object from Kafka?</p>