Github user bowenli86 commented on a diff in the pull request:

    https://github.com/apache/flink/pull/4833#discussion_r146134312
  
    --- Diff: docs/dev/stream/operators/windows.md ---
    @@ -721,6 +808,111 @@ input
     </div>
     </div>
     
    +#### Incremental Window Aggregation with AggregateFunction
    +
    +The following example shows how an incremental `AggregateFunction` can be 
combined with
    +a `ProcesWindowFunction` to compute the average and also emit the key and 
window along with
    +the average.
    +
    +<div class="codetabs" markdown="1">
    +<div data-lang="java" markdown="1">
    +{% highlight java %}
    +DataStream<Tuple2<String, Long> input = ...;
    +
    +input
    +  .keyBy(<key selector>)
    +  .timeWindow(<window assigner>)
    +  .aggregate(new AverageAggregate(), new MyProcessWindowFunction());
    +
    +// Function definitions
    +
    +/**
    + * The accumulator is used to keep a running sum and a count. The {@code 
getResult} method
    + * computes the average.
    + */
    +private static class AverageAggregate
    +    implements AggregateFunction<Tuple2<String, Long>, Tuple2<Long, Long>, 
Double> {
    +  @Override
    +  public Tuple2<Long, Long> createAccumulator() {
    +    return new Tuple2<>(0L, 0L);
    +  }
    +
    +  @Override
    +  public Tuple2<Long, Long> add(
    +    Tuple2<String, Long> value, Tuple2<Long, Long> accumulator) {
    +    return new Tuple2<>(accumulator.f0 + value.f1, accumulator.f1 + 1L);
    +  }
    +
    +  @Override
    +  public Double getResult(Tuple2<Long, Long> accumulator) {
    +    return accumulator.f0 / accumulator.f1;
    +  }
    +
    +  @Override
    +  public Tuple2<Long, Long> merge(
    +    Tuple2<Long, Long> a, Tuple2<Long, Long> b) {
    --- End diff --
    
    ditto


---

Reply via email to