Flink DataStream Manager(托管) Operator State的简单使用

雨点打透心脏的1/2处 2021-12-18 14:47 300阅读 0赞

## Flink DataStream Manager(托管) Operator State的简单使用 ##

要使用Manager(托管) `Operator State`，需要实现`CheckpointedFunction`接口或者`ListCheckpointed<T extends Serializable>`接口。

#### 一、CheckpointFunction ####

`CheckpointedFunction`接口提供具有不同重新分发`Non-Keyed State`的访问状态。他需要实现2个方法：

void snapshotState(FunctionSnapshotContext context) throws Exception;
    
    void initializeState(FunctionInitializationContext context) throws Exception;

每当必须执行checkpoint时，都会有`snapshotState()`方法调用。`initializeState()`方法首次初始化函数或从早期的checkpoint中恢复时被调用。

目前，支持列表样式的托管Operator State。状态被预期是一个`List`的*序列化的*对象，彼此独立的，因而在重新缩放获再分配。换句话说，这些对象是可以重新分配非键控状态的最精细的粒度。根据状态访问方法，定义了以下重新分发方案：

*  \*\*偶分裂再分配：\*\*每个运算符返回一个状态元素列表。整个状态在逻辑上是所有列表的串联。在恢复/重新分发时，列表被平均分成与并行运算符一样多的子列表。每个运算符都获得一个子列表，该子列表可以为空，或包含一个或多个元素。例如，如果使用并行性1，则运算符的检查点状态包含元素`element1`和`element2`，当将并行性增加到2时，`element1`可能最终在运算符实例0分区中，而`element2`将转到运算符实例1分区。
 *  \*\*联合重新分配：\*\*每个运算符返回一个状态元素列表。整个状态在逻辑上是所有列表的串联。在恢复/重新分配时，每个运算符都会获得完整的状态元素列表。

下面是一个有状态的示例`SinkFunction`，用于`CheckpointedFunction` 在将元素发送到外部世界之前对其进行缓冲。它演示了基本的偶分裂再分发列表状态：

public class BufferingSink implements SinkFunction<Tuple2<String, Integer>>, CheckpointedFunction { 
    
        private final int threshold;
    
        private transient ListState<Tuple2<String, Integer>> checkpointedState;
    
        private List<Tuple2<String, Integer>> bufferedElements;
    
        public BufferingSink(int threshold) { 
            this.threshold = threshold;
            this.bufferedElements = new ArrayList<>();
        }
    
        @Override
        public void invoke(Tuple2<String, Integer> value, Context contex) throws Exception { 
            bufferedElements.add(value);
            if (bufferedElements.size() == threshold) { 
                for (Tuple2<String, Integer> element : bufferedElements) { 
                    // send it to the sink
                }
                bufferedElements.clear();
            }
        }
    
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception { 
            checkpointedState.clear();
            for (Tuple2<String, Integer> element : bufferedElements) { 
                checkpointedState.add(element);
            }
        }
    
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception { 
            ListStateDescriptor<Tuple2<String, Integer>> descriptor = new ListStateDescriptor<>("buffered-elements", TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() { 
            }));
    
            checkpointedState = context.getOperatorStateStore().getListState(descriptor);
    
            if (context.isRestored()) { // isRestored：检查是否是失败后恢复
                for (Tuple2<String, Integer> element : checkpointedState.get()) { 
                    bufferedElements.add(element);
                }
            }
        }
    }

如果要在还原时使用**联合重新分发**方案的列表状态，可以使用`getUnoinListState(descriptor)`方法获取。如果是`getListState(descriptor)`，它仅表示将使用基本**偶分裂再分配**方案。

#### 二、ListCheckpointed ####

该`ListCheckpointed`接口是CheckpointedFunction的变体。**它仅支持与恢复甚至分裂的再分配方案**列表式的状态。它还需要实现两种方法：

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

在`snapshotState()`操作人员应对象返回检查点的列表，并 `restoreState`具有处理后恢复这样的列表。如果状态不是重新分区，可以随时返回`Collections.singletonList(MY_STATE)`的`snapshotState()`。

下面使用自定义SourceFunction实现ListCheckpointed接口：

public class CustomSourceFunction extends RichParallelSourceFunction<Long> implements ListCheckpointed<Long> { 
    
        /** * current offset for exactly once semantics */
        private Long offset = 0L;
    
        /** * flag for job cancellation */
        private volatile boolean isRunning = true;
    
        @Override
        public void run(SourceContext<Long> ctx) { 
            final Object lock = ctx.getCheckpointLock();
    
            while (isRunning) { 
                // output and state update are atomic
                synchronized (lock) { 
                    ctx.collect(offset);
                    offset += 1;
                }
            }
        }
    
        @Override
        public void cancel() { 
            isRunning = false;
        }
    
        @Override
        public List<Long> snapshotState(long checkpointId, long checkpointTimestamp) { 
            return Collections.singletonList(offset);
        }
    
        @Override
        public void restoreState(List<Long> state) { 
            for (Long s : state) { 
                offset = s;
            }
        }
    }