flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-5968) Document WindowedStream.aggregate()
Date Sun, 22 Oct 2017 15:05:01 GMT

    [ https://issues.apache.org/jira/browse/FLINK-5968?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16214338#comment-16214338
] 

ASF GitHub Bot commented on FLINK-5968:
---------------------------------------

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


> Document WindowedStream.aggregate()
> -----------------------------------
>
>                 Key: FLINK-5968
>                 URL: https://issues.apache.org/jira/browse/FLINK-5968
>             Project: Flink
>          Issue Type: Sub-task
>          Components: DataStream API, Documentation
>            Reporter: Aljoscha Krettek
>            Assignee: Aljoscha Krettek
>            Priority: Blocker
>             Fix For: 1.4.0
>
>




--
This message was sent by Atlassian JIRA
(v6.4.14#64029)

Mime
View raw message