public class Main {

  /**
   * (1,4)
   * (1,6)
   */
  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.fromElements(Tuple2.of(1, 3), Tuple2.of(1, 5), Tuple2.of(1, 7), Tuple2.of(1, 5), Tuple2.of(1, 2))
            .keyBy(x->x.f0)
            .flatMap(new CountWindowAverage())
            .print();
    env.execute("test");
  }

  private static class CountWindowAverage extends RichFlatMapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>> {

    private ValueState<Tuple2<Integer, Integer>> state = null;

    @Override
    public void open(Configuration parameters) {
      ValueStateDescriptor<Tuple2<Integer, Integer>> descriptor =
              new ValueStateDescriptor<>("average", TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {
              }));

      // 设置ttl
      StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.seconds(10))
              // 何时更新过期状态
              .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
              // 是否可以访问过期数据
              .setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
              .build();
      descriptor.enableTimeToLive(ttlConfig);

      state = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(Tuple2<Integer, Integer> value, Collector<Tuple2<Integer, Integer>> out) throws Exception {
      Tuple2<Integer, Integer> curSum = Optional
              .ofNullable(state.value())
              .orElse(Tuple2.of(0, 0));

      curSum.f0 += 1;
      curSum.f1 += value.f1;
      state.update(curSum);

      if (curSum.f0 >= 2) {
        out.collect(Tuple2.of(value.f0, curSum.f1 / curSum.f0));
        state.clear();
      }
    }
  }
}