Flink 实时计算入门：从 WordCount 到状态管理

StreamExecutionEnvironment env =
        StreamExecutionEnvironment.getExecutionEnvironment();

env.socketTextStream("localhost", 9999)
   .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
       @Override
       public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
           for (String word : value.split("\\s")) {
               out.collect(new Tuple2<>(word, 1));
           }
       }
   })
   .keyBy(t -> t.f0)
   .sum(1)
   .print();

env.execute("Streaming WordCount");

WatermarkStrategy<Event> strategy = WatermarkStrategy
        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
        .withTimestampAssigner((event, ts) -> event.getTimestamp());

env.addSource(kafkaSource)
   .assignTimestampsAndWatermarks(strategy)
   .keyBy(Event::getUserId)
   .window(TumblingEventTimeWindows.of(Time.minutes(1)))
   .aggregate(new CountAgg())
   .print();

public class CountWithState extends RichFlatMapFunction<Event, Tuple2<String, Long>> {

    private transient ValueState<Long> countState;

    @Override
    public void open(Configuration conf) {
        ValueStateDescriptor<Long> desc =
                new ValueStateDescriptor<>("count", Long.class);
        countState = getRuntimeContext().getState(desc);
    }

    @Override
    public void flatMap(Event e, Collector<Tuple2<String, Long>> out) throws Exception {
        Long current = countState.value();
        if (current == null) current = 0L;
        current++;
        countState.update(current);
        out.collect(new Tuple2<>(e.getUserId(), current));
    }
}

env.enableCheckpointing(60000);   // 每 60s
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setCheckpointTimeout(600000);
env.setStateBackend(new RocksDBStateBackend("hdfs://namenode/flink/checkpoints"));