Page History

...

When using timestamp based windows the timestamp that get's attached to the result of a windowing operation becomes important. For example, because the timestamp is used in downstream window operations (as we will see below). The timestamp of the result of a window operation is always the lowest timestamp among the elements that make up the window. For example, in this reduce operation this can be seen as folding all elements into the first element (which has the lowest timestamp), thereby keeping the timestamp of this first element.

val stream: DataStream[MyType] = env.createStream(new KafkaSource(...))
val windowed: DataStream[MyType] = stream
  .keyBy(...)
  .window(...)
  .reduce { ... }

...