Flink 的状态管理和检查点机制

1
2
3
4
5
6

//在创建检查点的时候调用
void snapshotState(FunctionSnapshotContext context) throws Exception;

//在初始化的时候调用 (在从检查点恢复状态的时候也会先调用该方法)
//通过 FunctionInitializationContext 可以访问到 OperatorStateStore 和 KeyedStateStore
void initializeState(FunctionInitializationContext context) throws Exception;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

public class MyFunction<T> implements MapFunction<T, T>, CheckpointedFunction {

     private ReducingState<Long> countPerKey;
     private ListState<Long> countPerPartition;

     private long localCount;

     public void initializeState(FunctionInitializationContext context) throws Exception {
         // get the state data structure for the keyed state
         countPerKey = context.getKeyedStateStore().getReducingState(
                 new ReducingStateDescriptor<>("perKeyCount", new AddFunction<>(), Long.class));

         // get the state data structure for the operator state
         countPerPartition = context.getOperatorStateStore().getListState(
                 new ListStateDescriptor<>("perPartitionCount", Long.class));

         // initialize the "local count variable" based on the operator state
         for (Long l : countPerPartition.get()) {
             localCount += l;
         }
     }

     public void snapshotState(FunctionSnapshotContext context) throws Exception {
         // the keyed state is always up to date anyways
         // just bring the per-partition state in shape
         countPerPartition.clear();
         countPerPartition.add(localCount);
     }

     public T map(T value) throws Exception {
         // update the states
         countPerKey.add(1L);
         localCount++;

         return value;
     }
 }

1
2
3

List<T> snapshotState(long checkpointId, long timestamp) throws Exception;

void restoreState(List<T> state) throws Exception;

1
2
3
4

ValueState<T> getState(ValueStateDescriptor<T>)
ReducingState<T> getReducingState(ReducingStateDescriptor<T>)
ListState<T> getListState(ListStateDescriptor<T>)
MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV>)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

public class CountWindowAverage extends RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {

    /**
     * The ValueState handle. The first field is the count, the second field a running sum.
     */
    private transient ValueState<Tuple2<Long, Long>> sum;

    @Override
    public void flatMap(Tuple2<Long, Long> input, Collector<Tuple2<Long, Long>> out) throws Exception {

        // access the state value
        Tuple2<Long, Long> currentSum = sum.value();

        // update the count
        currentSum.f0 += 1;

        // add the second field of the input value
        currentSum.f1 += input.f1;

        // update the state
        sum.update(currentSum);

        // if the count reaches 2, emit the average and clear the state
        if (currentSum.f0 >= 2) {
            out.collect(new Tuple2<>(input.f0, currentSum.f1 / currentSum.f0));
            sum.clear();
        }
    }

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
                new ValueStateDescriptor<>(
                        "average", // the state name
                        TypeInformation.of(new TypeHint<Tuple2<Long, Long>>() {}), // type information
                        Tuple2.of(0L, 0L)); // default value of the state, if nothing was set
        sum = getRuntimeContext().getState(descriptor);
    }
}

// this can be used in a streaming program like this (assuming we have a StreamExecutionEnvironment env)
env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(1L, 4L), Tuple2.of(1L, 2L))
        .keyBy(0)
        .flatMap(new CountWindowAverage())
        .print();

// the printed output will be (1,4) and (1,5)