Flink Operator State 实例实现CheckpointedFunction

亦凉 2021-09-24 11:40 544阅读 0赞

public interface CheckpointedFunction {
      
      // Checkpoint触发时会调用这个方法，我们要实现具体的snapshot逻辑，比如将哪些本地状态持久化
      void snapshotState(FunctionSnapshotContext context) throws Exception;
      // 第一次初始化时会调用这个方法，或者从之前的检查点恢复时也会调用这个方法
      void initializeState(FunctionInitializationContext context) throws Exception;
    }

可以自定义把哪些变量如何保存到检查点，以及如何初始化、如何从检查点恢复

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 per-key state
              countPerKey = context.getKeyedStateStore().getReducingState(
                      new ReducingStateDescriptor<>("perKeyCount", new AddFunction<>(), Long.class));
     
              // get the state data structure for the per-partition state
              countPerPartition = context.getOperatorStateStore().getOperatorState(
                      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;
          }
       }

flink 1.9.2, java1.8