title: How to aggregate events in flink stream before merging with current state by reduce function?
body: <p>My events are like: <code>case class Event(user: User, stats: Map[StatType, Int])</code></p>
  <p>Every event contains +1 or -1 values in it.
  I have my current pipeline that works fine but produces new event for every change of statistics.</p>
  <pre class="lang-scala prettyprint-override"><code>eventsStream
      .keyBy(extractKey)
      .reduce(reduceFunc)
      .map(prepareRequest)
      .addSink(sink)
  </code></pre>
  <p>I'd like to aggregate these increments in a time window before merging them with the current state. So I want the same rolling reduce but with a time window.</p>
  <p>Current simple rolling reduce:</p>
  <pre><code>500 – last reduced value
  +1
  -1
  +1
  
  Emitted events: 501, 500, 501 
  </code></pre>
  <p>Rolling reduce with a window:</p>
  <pre><code>500 – last reduced value
  v-- window
  +1
  -1
  +1
  ^-- window
  
  Emitted events: 501
  </code></pre>
  <p>I've tried naive solution to put time window just before <code>reduce</code> but after reading the docs I see that reduce now has different behavior.</p>
  <pre class="lang-scala prettyprint-override"><code>eventsStream
      .keyBy(extractKey)
      .timeWindow(Time.minutes(2))
      .reduce(reduceFunc)
      .map(prepareRequest)
      .addSink(sink)
  </code></pre>
  <p>It seems that I should make keyed stream and reduce it after reducing my time window:</p>
  <pre class="lang-scala prettyprint-override"><code>eventsStream
      .keyBy(extractKey)
      .timeWindow(Time.minutes(2))
      .reduce(reduceFunc)
      .keyBy(extractKey)
      .reduce(reduceFunc)
      .map(prepareRequest)
      .addSink(sink)
  </code></pre>
  <p>Is it the right pipeline to solve a problem?</p>