title: Calculate totals and emit periodically in flink
body: <p>I have a stream of events about resources that looks like this:</p>
  
  <pre><code>id, type,      count
  1,  view,      1
  1,  download,  3
  2,  view,      1
  3,  view,      1
  1,  download,  2
  3,  view,      1
  </code></pre>
  
  <p>I am trying to produce stats (totals) per resource, so if I get a stream like above, the result should be:</p>
  
  <pre><code>id, views, downloads
  1,  1,     5
  2,  1,     0
  3,  2,     0
  </code></pre>
  
  <p>Now I wrote a ProcessFunction that calculates the totals like this:</p>
  
  <pre><code>public class CountTotals extends ProcessFunction&lt;Event, ResourceTotals&gt; {
      private ValueState&lt;ResourceTotals&gt; totalsState;
  
      @Override
      public void open(Configuration config) throws Exception {
          ValueStateDescriptor&lt;ResourceTotals&gt; totalsDescriptor = new ValueStateDescriptor&lt;&gt;("totals state", ResourceTotals.class);
          totalsDescriptor.setQueryable("resource-totals");
          totalsState = getRuntimeContext().getState(totalsDescriptor);
      }
  
      @Override
      public void processElement(Event event, Context ctx, Collector&lt;ResourceTotals&gt; out) throws Exception {
          ResourceTotals totals = totalsState.value();
          if (totals == null) {
              totals = new ResourceTotals();
              totals.id = event.id;
          }
          switch (event.type) {
              case "view":
                  totals.views += event.count;
                  break;
              case "download":
                  totals.downloads += event.count;
          }
          totalsState.update(totals);
          out.collect(totals);
      }
  }
  </code></pre>
  
  <p>As evident from the code, it will emit a new ResourceTotals for every event, but I would like to emit a total per resource once a minute and not more often.</p>
  
  <p>I tried experimenting by using a global window and a trigger (ContinuousProcessingTimeTrigger), but couldn't get it to work. The issues I had with it were:</p>
  
  <ol>
  <li>How to express that I want the last event of the window?</li>
  <li>How not to end up storing all the ResourceTotals ever produced in that global window?</li>
  </ol>
  
  <p>Any help would be appreciated.</p>