接口介绍

见名知意，就是ck的底层接口。一般用于您需要使用应该由Flink管理并包含在检查点中的状态，但是您不使用KeyedStream，因此不能像在KeyedProcessFunction中那样使用键控状态。

public interface CheckpointedFunction {


	// 当每次任务触发checkpoint时执行，更新保存状态数据
	void snapshotState(FunctionSnapshotContext context) throws Exception;

	// 初始化checkpoint 存储结构，一般在这里我们会实现两个逻辑：
        // 1.判断checkpoint 是否是重启状态恢复，并实现状态恢复逻辑
        // 2.初始化checkpoint存储逻辑规则。
	void initializeState(FunctionInitializationContext context) throws Exception;
}

示例一

场景

在ProcessFunction中使用，用于计算作业重新启动了多少次。

代码

public class MyProcessFunction<V> extends ProcessFunction<V, V> implements CheckpointedFunction {
    ...

    private transient ListState<Long> restarts;

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception { ... }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        restarts = context.getOperatorStateStore().getListState(new ListStateDescriptor<Long>("restarts", Long.class));

        if (context.isRestored()) {
            List<Long> restoreList = Lists.newArrayList(restarts.get());
            if (restoreList == null || restoreList.isEmpty()) {
                restarts.add(1L);
                System.out.println("restarts: 1");
            } else {
                Long max = Collections.max(restoreList);
                System.out.println("restarts: " + max);
                restarts.add(max + 1);
            }
        } else {
            System.out.println("restarts: never restored");
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception { ... }

    @Override
    public void processElement(V value, Context ctx, Collector<V> out) throws Exception { ... }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<V> out) throws Exception { ... }

示例二

场景

在RichFlatMapFunction中使用，用于攒批

代码

//定义一个抽象类
public abstract class BulkProcessFlatMap extends RichFlatMapFunction<String, List<String>> implements CheckpointedFunction {

    private static final Logger logger = LoggerFactory.getLogger(BulkProcessFlatMap.class);
    private transient ListState<String> checkpointedState;
    private List<String> bufferedElements;
    private BulkContext bulkContext;
    private Collector<List<String>> out;

    public BulkProcessFlatMap() {
        this.bufferedElements = new ArrayList<>();
    }

    @Override
    public void open(Configuration parameters) throws Exception {

        bulkContext = getBulkContext();
    }

    @Override
    public void flatMap(String value, Collector<List<String>> out) throws Exception {

        bufferedElements.add(value);
        if (bufferedElements.size() == bulkContext.getBatchSize() || System.currentTimeMillis() - bulkContext.getLastModified() >= bulkContext.getCacheTime()) {
            out.collect(bufferedElements);
            bufferedElements.clear();
            bulkContext.setLastModified(System.currentTimeMillis());
        }
        this.out = out;
    }

    @Override
    public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {

        checkpointedState.clear();
        if (ListUtils.isEmpty(bufferedElements)) {
            return;
        }
        long startTime = System.currentTimeMillis();
        if (this.out != null && startTime - bulkContext.getLastModified() >= bulkContext.getTimeOut()) {
            this.out.collect(bufferedElements);
            logger.info("Bulk Process FaltMap 超时输出成功:{}|{}ms", bufferedElements.size(), System.currentTimeMillis() - startTime);
            bufferedElements.clear();
        } else {
            checkpointedState.addAll(bufferedElements);
            logger.info("Bulk Process FaltMap State备份数据成功:{}|{}ms", bufferedElements.size(), System.currentTimeMillis() - startTime);
        }
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {

        long startTime = System.currentTimeMillis();
        ListStateDescriptor<String> descriptor = new ListStateDescriptor<>("buffered-list", TypeInformation.of(new TypeHint<String>() {}));
        checkpointedState = context.getOperatorStateStore().getListState(descriptor);
        if (context.isRestored()) {
            for (String element : checkpointedState.get()) {
                bufferedElements.add(element);
            }
            logger.info("Bulk Process FaltMap State初始化恢复数据成功:{}|{}ms", bufferedElements.size(), System.currentTimeMillis() - startTime);
        }
    }

    public abstract BulkContext getBulkContext();
}




//实现类
public class MyBulkProcessFlatMap extends BulkProcessFlatMap {

    @Override
    public BulkContext getBulkContext() {

        BulkContext bulkContext = new BulkContext();
        bulkContext.setBatchSize(100);
        bulkContext.setCacheTime(30 * 1000L);
        return bulkContext;
    }
}