Flink Operator State 实例实现ListCheckpointed

墨蓝 2021-09-24 11:40 249阅读 0赞

public interface ListCheckpointed<T extends Serializable> {
      // Checkpoint触发时会调用这个方法，我们要实现具体的snapshot逻辑，比如将哪些本地状态持久化
      List<T> snapshotState(long checkpointId, long timestamp) throws Exception;
      // 从上次Checkpoint中恢复数据到本地内存
      void restoreState(List<T> state) throws Exception;
    }

`ListCheckpointed`接口类是`CheckpointedFunction`接口类的一种简写，`ListCheckpointed`提供的功能有限，只支持均匀分布的ListState，不支持全量广播的UnionListState。

跟`CheckpointedFunction`中的`snapshotState`方法一样，这里的`snapshotState`也是在做备份，但这里的参数列表更加精简，其中`checkpointId`是一个单调递增的数字，用来表示某次Checkpoint，`timestamp`是Checkpoint发生的实际时间，这个方法以列表形式返回需要写入存储的状态。`restoreState`方法用来初始化状态，包括作业第一次启动或者作业失败重启。参数是一个列表形式的状态，是均匀分布给这个算子子任务的状态数据。

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

public class CountingFunction<T> implements MapFunction<T, Tuple2<T, Long>>, ListCheckpointed<Long> {
     
         // this count is the number of elements in the parallel subtask
         private long count;
     
         @Override
         public List<Long> snapshotState(long checkpointId, long timestamp) {
             // return a single element - our count
             return Collections.singletonList(count);
         }
     
         @Override
         public void restoreState(List<Long> state) throws Exception {
             // in case of scale in, this adds up counters from different original subtasks
             // in case of scale out, list this may be empty
             for (Long l : state) {
                 count += l;
             }
         }
     
         @Override
         public Tuple2<T, Long> map(T value) {
             count++;
             return new Tuple2<>(value, count);
         }
     }

flink 1.9.2, java1.8