title: Flink windowing: aggregate and output to sink
body: <p>We have a stream of data where each element is of this type:</p>
  
  <pre><code>id: String
  type: Type
  amount: Integer
  </code></pre>
  
  <p>We want to aggregate this stream and output the sum of <code>amount</code> once per week.</p>
  
  <p><strong>Current solution:</strong></p>
  
  <p>A example flink pipeline would look like this:</p>
  
  <pre><code>stream.keyBy(type)
        .window(TumblingProcessingTimeWindows.of(Time.days(7)))
        .reduce(sumAmount())
        .addSink(someOutput())
  </code></pre>
  
  <p>For input</p>
  
  <pre><code>| id | type | amount |
  | 1  | CAT  | 10     |
  | 2  | DOG  | 20     |
  | 3  | CAT  | 5      |
  | 4  | DOG  | 15     |
  | 5  | DOG  | 50     |
  </code></pre>
  
  <p>if the window ends between record <code>3</code> and <code>4</code> our output would be:</p>
  
  <pre><code>| TYPE | sumAmount |
  | CAT  | 15        | (id 1 and id 3 added together)
  | DOG  | 20        | (only id 2 as been 'summed')
  </code></pre>
  
  <p>Id <code>4</code> and <code>5</code> would still be inside the flink pipeline and will be outputted next week. </p>
  
  <p>Thus next week our total output would be:</p>
  
  <pre><code>| TYPE | sumAmount |
  | CAT  | 15        | (of last week)
  | DOG  | 20        | (of last week)
  | DOG  | 65        | (id 4 and id 5 added together)
  </code></pre>
  
  <p><strong>New requirement:</strong></p>
  
  <p>We now also want to know for each record in what week has each record been processed. In other words our new output should be:</p>
  
  <pre><code>| TYPE | sumAmount | weekNumber |
  | CAT  | 15        | 1          |
  | DOG  | 20        | 1          |
  | DOG  | 65        | 2          |
  </code></pre>
  
  <p>but we also want an additional output like this:</p>
  
  <pre><code>| id | weekNumber |
  | 1  | 1          |
  | 2  | 1          |
  | 3  | 1          |
  | 4  | 2          |
  | 5  | 2          |
  </code></pre>
  
  <p><strong>How to handle this?</strong></p>
  
  <p>Does flink have any way to achieve this? I would image we would have an aggregate function that sums the amounts but also outputs each record with the current week number for example but I don't find a way to do this in the docs.</p>
  
  <p>(Note: we process about a 100 million records a week, so ideally we would  only like to keep the aggregates in flink's state during the week, not all individual records)</p>
  
  <p><strong>EDIT:</strong></p>
  
  <p>I went for the solution described by Anton below:</p>
  
  <pre><code>DataStream&lt;Element&gt; elements = 
    stream.keyBy(type)
          .process(myKeyedProcessFunction());
  
  elements.addSink(outputElements());
  elements.getSideOutput(outputTag)
          .addSink(outputAggregates())
  </code></pre>
  
  <p>And the KeyedProcessFunction looks something like:</p>
  
  <pre><code>class MyKeyedProcessFunction extends KeyedProcessFunction&lt;Type, Element, Element&gt;
      private ValueState&lt;ZonedDateTime&gt; state;
      private ValueState&lt;Integer&gt; sum;
  
      public void processElement(Element e, Context c, Collector&lt;Element&gt; out) {
          if (state.value() == null) {
              state.update(ZonedDateTime.now());
              sum.update(0);
              c.timerService().registerProcessingTimeTimer(nowPlus7Days);
          }
          element.addAggregationId(state.value());
          sum.update(sum.value() + element.getAmount());
      }
  
      public void onTimer(long timestamp, OnTimerContext c, Collector&lt;Element&gt; out) {
          state.update(null);
          c.output(outputTag, sum.value()); 
      }
  } 
  </code></pre>