3. KeyedState只能用于RichFunction

   // KeyedProcessFunction继承自AbstractRichFunction
   class FraudDetectorV2 extends KeyedProcessFunction[Long, Transaction, Alert]

4. 将State声明为实例级别变量

   @transient private var flagState: ValueState[java.lang.Boolean] = _

5. 在实现自RichFunction的open方法中创建State描述符，并为State赋值

   /**
    * 应该在方法开始处理数据之前注册State
    * 所以在open()方法内使用ValueStateDescriptor注册state
    * ValueStateDescriptor包含了告诉flink如何管理变量的元数据
    *
    * open方法由TaskManager的StreamTask在invoke方法的beforeInvoke内调用，
    * 即在调用task主逻辑之前调用open方法一次
    */
    @throws[Exception]
    override def open(parameters: Configuration): Unit = {
      // 1. 创建State描述符 StateDescriptor
      // state名为flag；
      // state类型为BOOLEAN
      val flagDescriptor = new ValueStateDescriptor("flag", Types.BOOLEAN)
   
      // 2. 创建用ValueStateDescriptor定义的state
      // getState(StateDescriptor) 方法初次调用时就是注册；restore时就是获取checkpoint里面状态last值
      flagState = getRuntimeContext.getState(flagDescriptor)
   
      // 创建timerState
      val timerDescriptor = new ValueStateDescriptor("timer-state", Types.LONG)
      timerState = getRuntimeContext.getState(timerDescriptor)
    }

6. 读写State

   // Get the current state for the current key
   val lastTransactionWasSmall = flagState.value
   
   // set the flag to true
   flagState.update(true)
   
   // clean up state
   flagState.clear()

5.1.4 Operator State

5.1.4.1 概述

• Operator State又称为non-keyed state，每一个 operator state 都仅与一个 operator 的实例绑定。
• 常见的 operator state 是source state，例如记录当前 KafkaSource consumer 的 offset

Operator State 可用于所有算子，常用于 Source，例如 FlinkKafkaConsumer.

public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFunction<T> implements
        CheckpointListener,
        ResultTypeQueryable<T>,
        CheckpointedFunction

5.1.4.2 分类

• ListState
• UnionList State
• BroadState State

5.1.4.3 并行度改变时

• ListState
  
  均匀分配，即将原来的State中的元素按照round-robin的模式均匀分配给新的Task即可
• UnionList State
  
  先将所有 State 合并为全量 State ，再分发给每个task实例。由用户来决定恢复时怎么做，比如特定某个并发做特定某个事情。
• BroadState State
  
  此时本来每个Task实例的State就相同，新的每个task也拿一份即可。

5.1.4.4 API使用

需要自己实现 CheckpointedFunction 或 ListCheckpointed 接口.

5.1.4.4.1 例1

fromElements会调用FromElementsFunction的类，其中就使用了类型为 list state 的 operator state

5.1.4.4.1 例2

public abstract class FlinkKafkaConsumerBase<T> extends RichParallelSourceFunction<T> implements
        CheckpointListener,
        ResultTypeQueryable<T>,
        CheckpointedFunction {

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

        OperatorStateStore stateStore = context.getOperatorStateStore();

        this.unionOffsetStates = stateStore.getUnionListState(new ListStateDescriptor<>(OFFSETS_STATE_NAME,
            createStateSerializer(getRuntimeContext().getExecutionConfig())));

        if (context.isRestored()) {
            // 恢复时，从ListState中读出所有分区Offset信息放入Map准备恢复
            restoredState = new TreeMap<>(new KafkaTopicPartition.Comparator());

            // populate actual holder for restored state
            for (Tuple2<KafkaTopicPartition, Long> kafkaOffset : unionOffsetStates.get()) {
                restoredState.put(kafkaOffset.f0, kafkaOffset.f1);
            }

            LOG.info("Consumer subtask {} restored state: {}.", getRuntimeContext().getIndexOfThisSubtask(), restoredState);
        } else {
            LOG.info("Consumer subtask {} has no restore state.", getRuntimeContext().getIndexOfThisSubtask());
        }
    }

    @Override
    public final void snapshotState(FunctionSnapshotContext context) throws Exception {
        if (!running) {
            LOG.debug("snapshotState() called on closed source");
        } else {
            // 先将旧状态清理
            unionOffsetStates.clear();

            final AbstractFetcher<?, ?> fetcher = this.kafkaFetcher;
            if (fetcher == null) {
                // the fetcher has not yet been initialized, which means we need to return the
                // originally restored offsets or the assigned partitions
                for (Map.Entry<KafkaTopicPartition, Long> subscribedPartition : subscribedPartitionsToStartOffsets.entrySet()) {
                    unionOffsetStates.add(Tuple2.of(subscribedPartition.getKey(), subscribedPartition.getValue()));
                }

                if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
                    // the map cannot be asynchronously updated, because only one checkpoint call can happen
                    // on this function at a time: either snapshotState() or notifyCheckpointComplete()
                    pendingOffsetsToCommit.put(context.getCheckpointId(), restoredState);
                }
            } else {
                // 获取当前Consumer offset
                HashMap<KafkaTopicPartition, Long> currentOffsets = fetcher.snapshotCurrentState();

                if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
                    // the map cannot be asynchronously updated, because only one checkpoint call can happen
                    // on this function at a time: either snapshotState() or notifyCheckpointComplete()
                    pendingOffsetsToCommit.put(context.getCheckpointId(), currentOffsets);
                }

                // 将当前offset放入状态快照
                for (Map.Entry<KafkaTopicPartition, Long> kafkaTopicPartitionLongEntry : currentOffsets.entrySet()) {
                    unionOffsetStates.add(
                            Tuple2.of(kafkaTopicPartitionLongEntry.getKey(), kafkaTopicPartitionLongEntry.getValue()));
                }
            }

            if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
                // truncate the map of pending offsets to commit, to prevent infinite growth
                while (pendingOffsetsToCommit.size() > MAX_NUM_PENDING_CHECKPOINTS) {
                    pendingOffsetsToCommit.remove(0);
                }
            }
        }
    }

    // 这一轮checkpoint完成时
    @Override
    public final void notifyCheckpointComplete(long checkpointId) throws Exception {
        if (!running) {
            LOG.debug("notifyCheckpointComplete() called on closed source");
            return;
        }

        final AbstractFetcher<?, ?> fetcher = this.kafkaFetcher;
        if (fetcher == null) {
            LOG.debug("notifyCheckpointComplete() called on uninitialized source");
            return;
        }

        if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
            // only one commit operation must be in progress
            if (LOG.isDebugEnabled()) {
                LOG.debug("Consumer subtask {} committing offsets to Kafka/ZooKeeper for checkpoint {}.",
                    getRuntimeContext().getIndexOfThisSubtask(), checkpointId);
            }

            try {
                final int posInMap = pendingOffsetsToCommit.indexOf(checkpointId);
                if (posInMap == -1) {
                    LOG.warn("Consumer subtask {} received confirmation for unknown checkpoint id {}",
                        getRuntimeContext().getIndexOfThisSubtask(), checkpointId);
                    return;
                }

                // 获取并移除本次已完成的checkpoint
                @SuppressWarnings("unchecked")
                Map<KafkaTopicPartition, Long> offsets =
                    (Map<KafkaTopicPartition, Long>) pendingOffsetsToCommit.remove(posInMap);

                // remove older checkpoints in map
                for (int i = 0; i < posInMap; i++) {
                    pendingOffsetsToCommit.remove(0);
                }

                if (offsets == null || offsets.size() == 0) {
                    LOG.debug("Consumer subtask {} has empty checkpoint state.", getRuntimeContext().getIndexOfThisSubtask());
                    return;
                }
                // 将这次状态快照记录的offset提交到kafka
                fetcher.commitInternalOffsetsToKafka(offsets, offsetCommitCallback);
            } catch (Exception e) {
                if (running) {
                    throw e;
                }
                // else ignore exception if we are no longer running
            }
        }
    }
}

5.1.5 State持久化

5.1.5.1概述

Flink通过联合使用StreamReplay和Checkpoint来实现容错。Checkpoint需要使用持久化存储来保存状态：

• 可回放重复消费记录的数据源
  比如消息队列（如Kafka）、文件系统（如HDFS）
• 可存放状态的持久化存储
  HDFS等分布式文件系统

其中每次Checkpoint就是标记每个输入流中的特定点以及每个运算符的对应状态。通过恢复算子状态以及从Checkpoint开始重放数据记录，可以恢复数据流以及同时保持一致性（ExactlyOnce处理语义）。

Flink容错机制的具体做法就是不断给分布式数据流做Checkpoint快照，存放到可配的地方（通常是分布式文件系统中）。当由于如机器、网络、软件等错误导致程序出错时，就停止该数据流，并从最近一次成功的Checkpoint恢复和重启算子，输入流也被重置到该快照点。而在恢复过程中被重复消费的数据会被保证不造成影响。

而Checkpoint的间隔时间，就是一种在执行中用于Checkpoint的开销和恢复时间的权衡方法：

• 间隔越长，恢复时的记录数越多，恢复耗时更长
• 间隔越短，一定时间内用于Checkpoint的时间百分比越长，恢复耗时更短

5.1.5.2 KeyedState和OperatorState Checkpoint区别

状态从本质上来说，是Flink算子子任务的一种本地数据，为了保证数据可恢复性，使用Checkpoint机制来将状态数据持久化输出到存储空间上。状态相关的主要逻辑有两项：

• 将算子子任务本地内存数据在Checkpoint时将状态快照后持久化到状态后端StateBackend；
• 初始化或重启应用时，以一定的逻辑从StateBackend中读出对应状态数据，并转为算子子任务的本地内存数据。

注意：

• Keyed State对这两项内容做了更完善的封装，开发者可以开箱即用，也就是说只需要注册和使用这类状态，不需要管Checkpoint快照和恢复，Flink已经帮你做了。

• 对于Operator State来说，每个算子子任务管理自己的Operator State，或者说每个算子子任务上的数据流共享同一个状态，可以访问和修改该状态。

  但Flink的算子子任务上的数据在程序重启、横向伸缩等场景下不能保证百分百的一致性。换句话说，重启Flink应用后，某个数据流元素不一定会和上次一样，还能流入该算子子任务上。因此，我们需要根据自己的业务场景来设计snapshot和restore的逻辑。为了实现这两个步骤，Flink提供了最为基础的CheckpointedFunction接口类。

  public interface CheckpointedFunction {

  // Checkpoint时会调用这个方法，我们要实现具体的snapshot逻辑，比如将哪些本地状态持久化
  void snapshotState(FunctionSnapshotContext context) throws Exception;

  // 初始化时会调用这个方法，向本地状态中填充数据
  void initializeState(FunctionInitializationContext context) throws Exception;

  }

5.2 Checkpoint

可参考：

• Checkpointing使用
• Checkpoints部署和恢复

5.2.1 概述

5.2.1.1 Chandy Lamport

Chandy Lamport算法，可以想象就是对水管吹气，把里面的水全部吹出来，此时就是管道水排空时系统状态。

对应到Flink，就是CheckpointCoordinator定时对Source发送Checkpoint Barr

5.2.1.2 怎么保证精确一次容错？

• 简单场景
  
  每条数据处理后就记录下该数据位置对应当时的状态。如果发生fail，就从最后一次成功的状态快照进行恢复，从而保证精确一次。
• 分布式场景
  
  

5.2.1.3 到底是什么的状态？

由于Flink中每个函数和算子都可以是有状态的，有状态函数可以在处理每个事件过程中存储数据，这使状态成为任何类型更复杂操作的关键构成。

Flink容错性的核心部分就是为分布式数据流和算子状态制作Checkpoint一致性快照，可以在出错时从Checkpoint中恢复状态和执行位置等。

相关详细内容还可参考

• 论文-Lightweight Asynchronous Snapshots for Distributed Dataflows
  描述Flink快照机制实现原理
• Chandy-Lamport algorithm
  Flink快照实现参照的算法

5.2.1.4 Checkpoint持久化了什么？

Checkpoint需要使用持久化存储来保存快照状态：

• 可回放重复消费记录的数据源
  比如消息队列（如Kafka）、文件系统（如HDFS）
• 可存放状态的持久化存储
  HDFS等分布式文件系统

状态快照是指Flink Job的全局一致性镜像，一个快照包括：

• 一个指向每个DataSource的指针（如Kafka的offset）
• 每个有状态算子的状态副本，该副本是处理了 sources offset 之前所有的事件后而生成的状态。

5.2.2 Checkpoint异步性

Checkpoint为异步执行的，Checkpoint屏障不会在lock步骤中传播，并且算子可以异步地为其状态制作快照。

Flink 的 StateBackend 利用写时复制（copy-on-write）机制允许当异步生成旧版本的状态快照时，能够不受影响地继续流处理。

只有当快照被持久保存到StateBackend后，这些旧版本的状态才会被当做垃圾回收。

5.2.3 Checkpoint Aligned和Unaligned

Flink1.11以前只支持对齐的Checkpoint，从1.11开始也可启用不对齐的Checkpoint了。

• 对齐的Checkpoint
  每个算子需要等到已接收所有上游发送的 Barrier 对齐后才可以进行 本算子状态的Snapshot ，完成后继续向后发送 Barrier。这样，在出现反压的情况下，Barrier 从上游算子传送到下游算子可能需要很长的时间，从而导致 Checkpoint 超时的问题。

• 非对齐的Checkpoint
  针对这一问题，Flink 1.11 新增了 Unaligned Checkpoint机制，开启后一旦收到第一个上游Barrier就可以开始执行Checkpoint，并把上下游之间正在传输的数据也作为状态保存到快照中，这样 Checkpoint 的完成时间大大缩短，不再依赖于算子的处理能力，解决了反压场景下 Checkpoint 可能超时的问题。

  可以通过 env.getCheckpointConfig().enableUnalignedCheckpoints();开启unaligned Checkpoint。

5.2.4 Checkpoint配置

val env = StreamExecutionEnvironment.getExecutionEnvironment()

• 开启Checkpoint
  Checkpoint默认关闭。需要用env.enableCheckpointing(n毫秒)来开启，这里的n指开启两次checkpoint之间的间隔毫秒数时间。

  env.enableCheckpointing(1000)

• exactly once / at least once 语义
  默认为exactly once:

  env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

• Checkpoint超时时间
  如果某次Checkpoint超过此阈值还没完成，则将进行中的Checkpoint干掉作废，单位毫秒

  env.getCheckpointConfig.setCheckpointTimeout(60000)

• 下次Checkpoint距离上一次Checkpoint完成后的最小时间间隔毫秒数。
  注意，此时不会管已设置的Checkpoint时间间隔和每次Checkpoint持续时间。也就是说，此时间隔时间永不会比本参数时间还小了。还有一点就是，此时并发Checkpoint数设置失效，强制设为1。

  env.getCheckpointConfig.setMinPauseBetweenCheckpoints(1500);

  比如，我们通过env.enableCheckpointing设定了是 2 个两次Checkpoint之间的间隔毫秒数时间为1000ms。那么如果一个Checkpoint耗时900ms，本来过100ms就应做下一个 Checkpoint，导致checkpoint过于频繁。这个时候，本设置让Checkpoint完成之后最少要等 500ms才开始下一个，可以防止 Checkpoint 太过于频繁而导致业务处理的速度下降。

  所以有时候因为某次Checkpoint时间过长，可能导致采用时间间隔方式受影响导致上一个Checkpoint刚完成没多久又开始下一个Checkpoint，可采用本方式。

• Checkpoint并发数
  默认为1，即只能有1个Checkpoint同时进行。一个没完另一个不会触发。减少对Chcekpoint开销，不会对系统产生过多影响。

  env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

  对于这种场景可以用并发Checkpoint：有确定的处理延迟（比如经常调用较好时的外部服务接口），但仍希望频繁Checkpoint，以期当发生错误时恢复很少的数据开始处理。

• Checkpoint失败后是否导致flink任务失败
  默认为true。

  env.getCheckpointConfig.setFailOnCheckpointingErrors(false)

  注：本方法已经废弃，建议改用以下方法，默认0即不容忍任何一个Checkpoint失败：

  env.getCheckpointConfig.setTolerableCheckpointFailureNumber(int)

• 是否开启外部检查点
  默认情况下，Checkpoint 在默认的情况下仅用于恢复失败的作业，而并不会被保留，即当程序Cancel时 Checkpoint 就会被删除。当然，你可以通过配置来保留 checkpoint，这些被保留的 checkpoint 在作业Fail或Cancel时不会被清除。

  我们可以将周期性的Checkpoint配置为保存到外部存储。这种检查点会将元数据持久化到外部存储，而且即使job失败也不会自动清除。这样，当job失败时可直接从这个现成的Checkpoint恢复。详细参考:Externalized checkpoints 的部署文档

  有两种选择：

  • ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
    当Cancel该job时也保留 Checkpoint。也就是说，我们必须在Cancel后需要手动删除Checkpoint文件。

  • ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION
    当Cancel该job时删除Checkpoint。仅当Job失败时，Checkpoint 才会被保留。

    为单个job设置Externalized方法：

    env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

  在flink-conf.yaml中全局设置方法：

  execution.checkpointing.externalized-checkpoint-retention:RETAIN_ON_CANCELLATION

• 优先从 Checkpoint 恢复而不是Savepoint
  默认为false。该属性确定 job 是否在最新的 checkpoint 回退，即使有更近的可以潜在地减少恢复时间的 savepoint 可用（因为checkpoint 恢复比 savepoint 恢复更快）

  env.getCheckpointConfig().setPreferCheckpointForRecovery(true);

5.2.5 Checkpoint原理

Checkpoint由运行在JM的组件CheckpointCoordinator定时触发：

// checkpointInterval
long baseInterval = chkConfig.getCheckpointInterval();
if (baseInterval < minPauseBetweenCheckpoints) {
    baseInterval = minPauseBetweenCheckpoints;
}

private ScheduledFuture<?> scheduleTriggerWithDelay(long initDelay) {
    return timer.scheduleAtFixedRate(
        new ScheduledTrigger(),
        // baseInterval就是我们配置的checkpointInterval
        initDelay, baseInterval, TimeUnit.MILLISECONDS);
}

private final class ScheduledTrigger implements Runnable {

    @Override
    public void run() {
        try {
            triggerCheckpoint(true);
        }
        catch (Exception e) {
            LOG.error("Exception while triggering checkpoint for job {}.", job, e);
        }
    }
}

5.2.5.1 CheckpointBarrier

分布式快照的核心组件被称为stream barrier，他们会被注入到数据流中和数据记录一起在算子之间流动，并且他们永不会超过记录，因为他们是按严格线性流动的。

Barrier将记录分为流入当前快照和下一个快照的两部分，且Barrier会记录下当前快照ID。Checkpoint n 将包含每个算子的状态，这些状态是对应的算子消费了严格在Checkpoint barrier n之前的所有事件，并且不包含在此后的任何事件后而生成的状态。

1. JM的CheckpointCoordinator定时触发Checkpoint流程

2. CheckpointBarrier被注入到Source阶段的并行数据流中，第n个快照的注入点Sn就是Source覆盖数据
   比如在Kafka中，注入点就是该partition最后一条记录的Offset。 第n个快照的注入点位置Sn会被报告给JobManager的组件CheckpointCoordinator。

   当Job 开始做Checkpoint barrier N 的时候可以理解为逐步将状态信息填充如下图左下角的表格。
   
   比如上图Source收到Barrier N后将partition Offset（比如Source当前消费Kafka分区Offset）放入Source状态表格中。

3. Barrier在注入后，和其他数据一起向数据流下游流动。
   

4. 当一个中间算子收到来自所有输入流的Snapshot N的对应Barrier N后，会将自己的状态异步写入持久化存储，并将Barrier N插入输出流发送给下游。

   Barrier N流动到Operator 1时，就将属于Checkpoint N到Checkpoint N-1之间的所有数据反映到当前状态快照中。
   

5. 一旦作为流程序DAG终点的某个Sink算子接收到所有输出流发来的Barrier N以后先填充自己的状态表格，随后就会认为快照N已经完成，此时会发送一个快照N已完成的ACK给CheckpointCoordinator。
   

6. 当CheckpointCoordinator收到所有Sink算子发出的ACK后，快照N就被认为已经成功执行完成。
   此后就不会再访问Sn以前的数据了，因为认为之前的数据已经走完整个数据拓扑。

7. 最后 CheckPointCoordinator 会把整个 StateHandle 封装成 completed CheckPoint Meta，写入到 hdfs。

8. 而如果直到Checkpoint超时，CheckPointCoordinator仍未收集完所有的 State Handle，CheckPointCoordinator会认为本次 CheckPoint 失败，将这次 CheckPoint 产生的所有状态数据全部删除。

5.2.5.2 Checkpoint Barrier对齐过程

当有多个输入流时，需要对齐输入流的同ID的Barrier。

当 job graph 中的每个 operator 都接收到该Barrier时，就会记录下自己的状态。

拥有两个输入流的 Operators（例如CoProcessFunction）会执行Checkpoint Barrier对齐（barrier alignment） ，以便当前快照能包含已消费两个输入流 barrier 之前（但不超过）的所有 event 而产生的状态。

1. 一旦算子从某个输入流收到BarrierN，那就不能再处理任何该输入流的数据，直到从其他所有输入流收到BarrierN。否则会导致将快照N和快照N+1的数据记录搞混。

   暂时停止处理的流的数据继续收到放入InputBuffer等待处理。

2. 一旦算子收集齐了BarrerN，就会立刻发送所有OutputBuffer中的PENDING状态的记录给下游，然后发送BarrierN。

3. 算子发送BarrierN完毕后，在内存中将状态制作快照，然后恢复从输入流获取、处理数据
   注意，会优先将InputBuffer中的数据处理完毕，然后再从输入流处理

4. 最后，算子将内存中制作好的状态异步写入StateBackend

注意，所有拥有多路输入流和消费来自多个上游子任务输出流的shuffle后的算子都需要使用Checkpoint对齐。

特点：

• 当若干快照并发执行时，数据流中会同时存在多个Barrier。
• 需要注意的是，其实Barrier不会中断数据流，所以非常轻量级。

5.2.5.3 算子状态快照

• 算子状态属于快照一部分
  只要你的算子包含任意格式的State，那就必须作为快照一部分保存。

• 算子状态快照时间点
  当算子接收到所有输出流的BarrierN后，就在内存中将状态制作快照，然后恢复从输入流获取、处理数据，最后算子将内存中制作好的状态异步写入StateBackend。

  在这个时间点上，所有在BarrierN之前的对State的更新已经做完，而在BarrierN之后对State的更新都还没做。

• StateBackend
  因为状态快照可能很大，不可能全放在内存，所以被存放到可配的StateBackend。

  默认在JM内存中，但生产中一般存放在分布式存储中，比如HDFS。

  当状态快照被存储后，算子会确认该次Checkpoint完成，发送该次快照的 Barrier到输出到下游的流中。

• 状态快照包含内容

  • 对于每个并行流Source，是快照启动时流中的offset/位置
  • 对于每个算子实例，是一个指向作为快照一部分的状态的指针

5.2.5.4 普通Checkpoint恢复

当失败时，直接从最近一次成功的Checkpoint恢复。

恢复时：

1. 重新部署整个分布式数据流
2. 将Checkpoint中包含的状态给Operator
3. Source会从Checkpoint中包含的位置K开始读取数据流
   比如Kafka中就是记录下的偏移量K。

5.2.5.5 增量Checkpoint恢复

先从最新的全量快照启动算子，然后将后续的增量快照更新到算子状态上，得到最新状态。

5.2.6 Unaligned Checkpoint

5.2.6.1 概述

Flink1.11以前只支持对齐的Checkpoint，从1.11开始也可启用不对齐的Checkpoint了。

• 对齐的Checkpoint
  每个算子需要等到已接收所有上游发送的 Barrier 对齐后才可以进行 本算子状态的Snapshot ，完成后继续向后发送 Barrier。这样，在出现反压的情况下，Barrier 从上游算子传送到下游算子可能需要很长的时间，从而导致 Checkpoint 超时的问题。

• 非对齐的Checkpoint
  针对这一问题，Flink 1.11 新增了 Unaligned Checkpoint机制，开启后一旦收到第一个上游Barrier就可以开始执行Checkpoint，并把上下游之间正在传输的数据也作为状态保存到快照中，这样 Checkpoint 的完成时间大大缩短，不再依赖于算子的处理能力，解决了反压场景下 Checkpoint 可能超时的问题。

  可以通过 env.getCheckpointConfig().enableUnalignedCheckpoints();开启unaligned Checkpoint。

特点：

• 优点
  CheckpointBarrier可以直接越过in-flight数据，所以与吞吐不再有关联
• 限制
  • 不能直接对非对齐Checkpoint Job进行rescale或修改JobGraph，必须在rescale前进行Savepoint(会对齐)
  • 目前不支持并发非对齐Checkpoint（本模式时间更短，所以其实不太需要并发）
  • 在恢复过程中，未对齐的检查点会中断有关水印的隐含保证，参考Unaligned checkpoints

5.2.6.2 原理

非对齐Checkpoint核心思想就是只要in-flight缓存数据会成为OperatorState的一部分，那Checkpoint就可以超越这些数据。

1. 算子收到第一个Barrier后，存到输入缓存InputBuffer中
2. 算子对缓存中的Barrier立刻处理，将该Barrier放入输出OutputBuffer末尾以立刻发送给下游算子
3. 算子将所有已超越的记录（上图中超过了 3、2、1、d、c、b、a、z、y、x）标记为异步存储，并据此创建其自身状态的快照。

因此，非对齐方式的Checkpoint算子仅短暂停止输入流处理以标记缓冲区、发送Barrier、创建其他状态的快照，不需要再像对齐Checkpoint那样等待所有所有输入流的Barrier。

非对齐Checkpoint可保证Barrier尽快到达Sink，所以特别适合某个流路径进展缓慢的情况，如果采用对齐Checkpoint甚至或延迟小时级别。

5.2.6.3 Unaligned Checkpoint恢复

1. 算子首先恢复in-flight数据
2. 然后再开始从上游算子消费处理数据，重新部署整个分布式数据流
3. 将Checkpoint中包含的状态给Operator
4. Source会从Checkpoint中包含的位置K开始读取数据流
   比如Kafka中就是记录下的偏移量K。

5.2.6.4 小结

• 使用场景
  • 多上游中的一个路径进展很慢
    非对齐方式的Checkpoint确保Barrier尽快到达Sink，特别适合至少有一个数据移动很缓慢的路径的应用，这种场景下对齐等待时间可能达到小时级。
  • 背压导致Checkpoint持续时间过长
    采用本方式后，大多数情况下Checkpoint持续时间和端到端延迟无关了
• 不适用场景
  但问题是会增加额外的IO压力，在写入StateBackend的IO是瓶颈时不适用。
• Savepoint只能对齐Checkpoint，不能非对齐Checkpoint。

5.2.7 Exactly Once / At Least Once / At Most Once

根据用户配置以及使用的集群，Flink有三种语义：

• Exactly Once
  数据既不丢失也不重复。注意这里的含义是保证引擎管理的状态更新只提交一次到持久的后端存储，而不是引擎只处理一次该数据。

  • Checkpoint
    Flink精确一次是由Checkpoint Barrier对齐实现的，通过这种机制，流应用程序中每个算子的所有状态都会定期一致性地做 checkpoint。一旦程序在运行中发生失败，就把所有算子都统一回滚到最后一次成功执行的全局一致Checkpoint。回滚期间，会STW。回滚后，Source源也回滚到了这个Checkpoint时的offset。也就是说，整个流应用程序基本上是回到最近一次的一致状态，然后程序可以从该状态重新启动。
    
    可以看到T3回滚后，Source offset已经重置，5,3会被重新处理，但他们仍然只会精确影响状态sum一次，所以结果依然正确。

  如果不需要可以设置CheckpointingMode.AT_LEAST_ONCE，此时可以提高性能，但不再是精确一次了。

    注意：Flink通过回退和重放Source数据流，Exactly Once并不意味着每个event只被精确处理一次，而是指每个event只会精确一次地影响由Flink管理的状态。

  • 至少一次事件传递和对重复数据去重
    对每个算子实现至少一次事件传递 + 数据去重，要求为每个算子维护一个事务日志
    

• At Least Once
  数据不丢，但可能重复

  具体做法是失败即从源头重试执行，但事件可能被处理多次。比如CheckpointBarrier不对齐时就进行快照，则可能导致先到达的那个流的数据被处理多次。
  

• At Most Once
  保证数据或事件最多由应用程序中的所有算子处理一次

  具体做法是不尝试从错误中恢复
  

5.2.8 Checkpoint Barrier对齐与Exactly Once 、 At Least Once

• At Least Once
  Checkpoint对齐有时候可能会增加大量耗时，对于需要超级低延时、但又要求一致性的程序，可以关掉对齐。也就是说，算子一旦看到Checkpoint Barrier就开始生成Checkpoint快照。

  当跳过Checkpoint对齐后，当一些输入流的BarrierN来到后，此时算子标记完成后可以继续处理所有输入，这就造成算子可能在CheckpointN完成之前又同时在处理属于CheckpointN+1的数据，因为这些后来的数据虽然在Checkpoint N Barrier之后，但会将其包含在这次Checkpoint备份的状态之中。

  这就造成在恢复CheckpointN时这些记录重复出现，因为他们既在CheckpointN的状态快照之中（已经影响了Checkpoint N记录的状态）又会在恢复时作为Checkpoint N之后数据的一部分被重放导致重复计算！

• Exactly Once
  所以要想Exactly Once，就要开启对齐，使用InputBuffer将对齐阶段继续接收的数据缓存，等待对齐完成后继续处理，此时如果从CheckpointN恢复时，自然其保存的状态就不会有CheckpointN后来的数据的干扰了。

注意：Checkpoint对齐只能用于拥有多个前驱结点的算子（如Join）和拥有多个后置节点的算子（repartitioning/shuaffle）。
所以只有一个后置或前驱结点的单并行算子（如map、flatMap、filter等）即使在at least once模式也会表现为exactly once。

Checkpoint Barrier对齐与不对齐例子可参考一文搞懂 Flink 的 Exactly Once 和 At Least Once

5.2.9 QA

转自 一文搞懂 Flink 的 Exactly Once 和 At Least Once，作者范瑞

5.3 Savepoint

5.3.1 概述

Savepoint 其实就是手动触发的 Checkpoint，它依靠常规的 checkpoint 机制获取程序的快照并将其写入StateBackend保存。

Savepoints 允许在不丢失任何状态的情况下升级程序和 Flink 集群。

5.3.2 Operator ID与Savepoint状态

Savepoint内部有一个类似map结构，为每个有状态算子保存了Operator ID -> State的映射。

所以我们使用DataStreamApi时，尽量为每个有状态算子声明ID，方便恢复。如果不指定会自动生成ID，只要这些ID不变就能手动从savepoint恢复。这些ID取决于程序，对程序改动很敏感。

正例如下：

DataStream<String> stream = env.
  // Stateful source (e.g. Kafka) with ID
  .addSource(new StatefulSource())
  .uid("source-id") // ID for the source operator
  .shuffle()
  // Stateful mapper with ID
  .map(new StatefulMapper())
  .uid("mapper-id") // ID for the mapper
  // Stateless printing sink
  .print(); // Auto-generated ID

此时savepoint生成的映射如下:

Operator ID | State
------------+------------------------
source-id   | State of StatefulSource
mapper-id   | State of StatefulMapper

可以看到，只生成了有状态算子的映射。而print这样的无状态算子被忽略，不会成为状态一部分。

5.3.3 Savepoint与Operator改动

• Savepoint后，在Job内增加了新有状态算子，再从Savepoint恢复
  可以，心有状态算子在没有状态情况下初始化
• Savepoint后，在Job内删除了有状态算子，再从Savepoint恢复
  此时会因为Savepoint默认会恢复状态内的所有算子给新Job，此时会导致无法恢复，只能加上allowNonRestoredState跳过无法恢复的状态
• Savepoint后，在Job内对有状态算子重新排序，再从Savepoint恢复
  如果有状态算子指定了uid，则不受影响；如果没有，则可能导致有状态算子自动生成的ID发生变化，导致无法恢复从以前的Savepoint恢复
• Savepoint后，在Job内对无状态算子添加、删除或重排序，再从Savepoint恢复
  如果有状态算子指定了uid，则不受影响；如果没有，则可能导致有状态算子自动生成的ID发生变化，导致无法恢复从以前的Savepoint恢复
• Savepoint后，改变程序并行度，再从Savepoint恢复
  没问题，恢复后使用新的并行度即可
• 可以移动savepoint内的文件吗？
  目前不行，因为metadata文件内使用的是绝对路径。非要改的话，就编辑该文件内容。

5.3.4 Savepoint对比Checkpoint

• Checkpoint在Flink Job执行期间定期在 TaskManager 节点上创建快照并生成 Checkpoint，所以在恢复时Flink 仅需要最后成功完成的Checkpoint即可。也就是说，一旦成功完成了新的Checkpoint，旧的就可以被丢弃。
• 管理者不同
  • Savepoints 由用户触发、管理、删除，并在新的 checkpoint 完成时不会自动过期，可通过命令行或在Cancel一个 Job 的同时通过 REST API 来创建 Savepoint。
  • Checkpoint由Flink负责管理、删除等。
• Checkpoint可以使用RocksDB增量模式，比Savepoint更轻量级
• Savepoint的数据以标准格式输出存储，允许版本升级或配置变更时使用

5.3.5 原理

将Checkpoint Barrier手动插入到所有Pipeline中从而产生分布式快照。

使用EventTime，所以每个事件总是放入同一个窗口，保证结果一致性。

5.4 State Backend 状态后端

可参考：

• State Backends

5.4.1 概述

Checkpoint 会将 timer 以及有状态算子中的状态进行一致性快照保存， 包括Connector（比如KafkaConnector、HDFSConnecotr等），Window以及任何用户自定义State。

DataStream API涉及到的State有：

• Window在触发计算前收集、聚合的元素
• 转换函数可能使用 key/value State 来存储值
• 转换继承CheckpointedFunction来 对存入State的值实现容错。
  比如StreamingFileSink实现该接口，使用状态存储了Bucket、Part-File，且实现了容错。

State Backend管理的状态包括:

• org.apache.flink.streaming.api.datastream.KeyedStream使用的Keyed State
• 实现org.apache.flink.streaming.api.checkpoint.CheckpointedFunction来直接管理的State

开启Checkpoint后，可以将上述State Checkpoint持久化，防止数据丢失，出错时可一致性恢复。而具体选择的StateBackend就决定了State内部如何组织表现，以及如何、在哪进行Checkpoint。

Checkpoint存储的可能位置包括JobManager内存、文件系统、数据库等（State默认存储在TaskManager内存中，而Checkpoint保存在JobManager内存中），具体取决于所配置的State Backend（比如持久化巨型状态就不适合放在内存），当前有以下State Backend可用于存储 Checkpoint State：

• MemoryStateBackend（默认）
• FsStateBackend
• RocksDBStateBackend
  
  

Job级别的StateBackend配置方式如下：

StreamExecutionEnvironment.setStateBackend()

这定义了该Job保存State的数据结构以及Checkpoint数据存储位置。

全局StateBackend配置方式如下：

• MemoryStateBackend为不配置时的默认选项。如果要修改默认配置，请修改flink-conf.yaml文件中的state.backend：

  state.backend: filesystem
  
  state.checkpoints.dir: hdfs://namenode:40010/flink/checkpoints

  其中：

  • state.backend
    可选内置的jobmanager (MemoryStateBackend), filesystem (FsStateBackend), rocksdb (RocksDBStateBackend)，或自己实现自StateBackendFactory的类全限定名，如org.apache.flink.contrib.streaming.state.RocksDBStateBackendFactory。

  • state.checkpoints.dir
    定义了StateBackend写Checkpoint数据和元数据文件路径。目录结构如下：

    /user-defined-checkpoint-dir
        /{job-id}
            |
            + --shared/
            + --taskowned/
            + --chk-1/
            + --chk-2/
            + --chk-3/
            ...

    shared目录保存了可能被多个Checkpoint引用的文件；taskowned存储JM不能删除的文件；chk开头的是每次checkpoint使用的文件，数字表示checkpointId。

5.4.2 MemoryStateBackend

• MemoryStateBackend为不配置时的默认选项。

• 存储位置

  • 平时
    以Java Object形式放在TaskManager的JVM内存中。具体来说， Key/value state 和 Window算子持有HashTable来存储State值和触发器等。
  • Checkpoint时
    将State进行快照，然后将它作为Checkpoint消息一部分发送给JobManager，然后放在JobManager的JVM内存中。

• 异步快照
  支持，且默认开启，强烈建议使用异步快照来防止数据流阻塞。

  要关闭异步采用同步方式（官方建议只在Debug时关闭异步Checkpoint）：

  new MemoryStateBackend(MAX_MEM_STATE_SIZE, false);

• 使用限制

  • 每个单独的State默认最大5MB，可用MemoryStateBackend构造函数设置

    env.setStateBackend(new MemoryStateBackend(MAX_MEM_STATE_SIZE, false))

  但是必须注意该参数谨慎调大，因为State Checkpoint时TaskManager会将State Checkpoint后的数据通过限制了大小的RPC方式发送给JobManager，而JobManager需要在内存保存来自各个TaskManager的状态数据。太大了可能导致OOM！

  • 同时由于Flink 节点间采用Akka通信，所以单状态大小受限于Akka Frame Size限制。
  • JobManager的JVM内存必须能存下所有聚合后的State

• 特点

  • 这种方式轻量级无需其他依赖，但受JVM内存限制，只能对小型State做Checkpoint，适用于
    • 本地开发测试
    • 计数器
    • 几乎无状态的Job，比如仅由每次只处理一条记录的算子（Map、FlatMap、Filter等）组成
    • KafkaConsumer，只需要很少的State用于记录消费offset情况
  • 每次运算值需要读取状态时使用Java object 读写，代价较小；

• 调优
  managed memory内存设为0，因为RocksDB StateBackend才会使用该部分内存。这样设定可以使得用户代码可以使用JVM能提供的最大值来执行。

5.4.3 FsStateBackend

• 存储位置

  • 平时
    以Java Object形式放在TaskManager的JVM内存中，读写代价小
  • Checkpoint发生时
    以Java Object形式将状态的Checkpoint序列化后保存到分布式文件系统，还有极小的State元数据存储在JobManager内存中（或是在高可用模式下将元数据存放在元数据Checkpoint中）。

• 异步快照
  支持，且默认开启异步快照Checkpoint来防止数据流阻塞。

  如果要关闭异步采用同步方式：

  new FsStateBackend(path, false);

  这里的Path是Checkpoint持久化目录，如hdfs://namenode:40010/flink/checkpoints 或file:///data/flink/checkpoints

• 特点

  • 当使用分布式文件系统如HDFS或Alluxio时，可保证单节点故障时也不会丢状态的数据，使得Flink程序高可用、强一致性。
  • 读写本地State时直接使用Java Object读写，开销较小
  • 当Checkpoint发生时需要将状态序列化再保存到HDFS，有一定开销。

• 适用场景

  • 大型State，很大的key/value State
  • 很宽的Window
  • 需要高可用
  • 同时对读写性能也有较高要求
  • 注意，不支持增量Checkpoint

• 调优
  managed memory内存设为0，因为RocksDB StateBackend才会使用该部分内存。这样设定可以使得用户代码可以使用JVM能提供的最大值来执行。

5.4.4 RocksDBStateBackend

可参考：

• RocksDB State Backend Details
• task executor memory configuration
• Tuning Checkpoints and Large State
• RocksDB Native Metrics
• 每个Slot中RocksDB实例占用的总内存大小配置

5.4.4.1 概述

• 简介
  RocksDB 是一个以日志合并树( LSM 树，Kudu、HBase都有使用）作为索引结构的 KV 存储引擎。当用于在 Flink 中存储 kv 状态时，Key由<Keygroup，Key，Namespace> 的序列化字节串组成，而Value由状态的序列化字节组成。

  每次注册 kv 状态时，它都会映射到列族（column-family），并将键值对以字节存储在 RocksDB 中。这意味着每次读写（READ or WRITE）操作都必须对数据进行反序列化或者序列化，与 Flink 内置的 in-memory 状态后端相比，会有一些性能开销。

• 存储位置

  • 平时
    将运行时State数据保存在RocksDB数据库中，RocksDB 数据库默认将数据存储在 TaskManager 节点磁盘上的的数据目录。
  • Checkpoint发生时
    将整个RocksDB数据库Checkpoint到配置的文件系统目录。同样的，还有极小的State元数据存储在JobManager内存中（或是在高可用模式下将元数据存放在元数据Checkpoint中）。

• 仅支持异步快照
  RocksDBStateBackend仅支持异步快照，且默认开启异步快照Checkpoint来防止数据流阻塞。

• 可支持增量Checkpoint
  RocksDB是唯一可支持的

• 使用限制
  由于 RocksDB 的 JNI API 构建在 byte[] 数据结构之上, 所以每个 key 和 value 最大支持2^31 字节。 注意: RocksDB 使用merge算子的状态（例如ListState）累积数据量大小可能悄悄超过 2^31 字节，会在下一次获取数据时失败。这是当前 RocksDB JNI 的限制。

• 特点

  • 可保留的State大小仅受可用磁盘空间量的限制。

    与平时将State保留在内存中的FsStateBackend相比，RocksDBStateBackend可以保留非常大的状态。 但这也意味着若使用RocksDBStateBackend，则可以实现的最大吞吐量将降低。

  • 每次读取状态都有序列化/反序列化开销
    从RocksDBStateBackend进行的所有读/写都必须经过序列化/反序列化以检索和存储State对象，这也比基于堆的StateBackend开销大很多。

  • 可使用增量快照

  • 不受 Java 垃圾回收的影响，与 heap 对象相比，它的内存开销更低

  • RocksDB是唯一可支持增量Checkpoint的

• 适用场景

  • 巨型State，巨大的key/value State
  • 很宽的Window，如天级窗口聚合计算
  • 需要高可用
  • 对计算性能要求较低，因为RocksDB读写有序列化/反序列化开销

5.4.4.2 内存管理

Flink花了很多功夫在内存管理上，以使得TM很好得利用内存，不至于在容器环境因为内存超限被杀掉，也不会因为内存利用率过低导致大量内存数据落盘或缓存命中率太低。

默认，RocksDB的可用内存配置为TM的一个Slot的内存量，一般用户无需调整细节，只需在不够时增加整体内存大小即可。state.backend.rocksdb.memory.managed就开启了RocksDB使用Managed Memory，Flink通过配置 RocksDB 来确保其使用的内存正好与 Flink 的Managed Memory预算相同，计算粒度是Per-Slot。也就是说，Flink会为一个Slot上的所有 RocksDB 实例使用共享的 RocksDB cache和 RocksDB write buffer manager

调整缓冲区：

• 写入缓冲内存不够时（现象为频繁flush）可调整state.backend.rocksdb.memory.write-buffer-ratio即分配给写入缓存的比例，默认0.5即50%。
• 读取缓冲命中率低时可调整state.backend.rocksdb.memory.high-prio-pool-ratio即分配给写入缓存的比例，默认0.1即10%，这部分内存优先分配给RocksDB的索引和过滤器。

但专业用户也可以手动为RocksDB每个列族（每个算子的每个State就对应了一个列族）分配内存，用户自己来确保总内存不会超限。

5.4.4.3 Timer（Heap vs RocksDB）

这里说的Timer是指用来触发窗口或回调ProcessFunction的类似操作，可基于事件时间或处理时间。

RocksDBStateBackend中Timer默认存在RocksDB，需要一定成本维护，所以也可以将Timer存在Java堆。

Timer较少时，可将state.backend.rocksdb.timer-service.factory设为heap，可获得更好的性能。但这样设置后，Timer状态就不能被异步快照存储了。

5.4.4.4 列族（ColumnFamily）级别的预定义选项

5.4.4.5 通过 RocksDBOptionsFactory 配置 RocksDB 选项

5.4.5 存储原理

5.4.5.1 概述

5.4.5.2 HeapKeyedStateBackend

对于HeapKeyedStateBackend，有两种实现：

• 支持异步 Checkpoint（默认）：存储格式CopyOnWriteStateMap
  

• 仅支持同步 Checkpoint：存储格式 NestedStateMap
  

• namespace用来标注如属于哪个window

• MemoryStateBackend 内使用HeapKeyedStateBackend时，Checkpoint 序列化数据阶段默认有最大 5 MB数据的限制
  

5.4.5.3 RocksDBKeyedStateBackend

• RocksDBKeyedStateBackend的每个 state 都存储在一个单独的 column family 内，使用了基于LSM树的磁盘、内存混合型DB
  

写入相关操作如下图

1. 将数据写入内存中的Active MemTable
2. MemTable会被后台线程周期性Flush到磁盘，生成按Key排序的只读的不可变文件SSTable
   当 MemTable 写满时，它将成为 READ ONLY MemTable，并被一个新申请的 MemTable 替换。
3. 刷到磁盘的 SSTable 文件会被后台线程多路归，并实现进一步的整合为大文件，提升读取效率
   合并后的 sstable 包含所有的键值对，RocksDB 会删除合并前的 sstable。
4. 每个注册状态都是一个列族，这意味着每个状态都包含独享的 MemTables 和 SSTables 集。

读取时：

• 首先访问Active MemTable。
• 若找不到，则访问SSTable
  • 优先查RocksDB BlockCache
  • 没有则查操作系统缓存PageCache
  • 在没有就查本地磁盘
  • 在此过程中可使用bloom filter减少大量磁盘访问，进行过滤

增量时（可参考Managing Large State in Apache Flink: An Intro to Incremental Checkpointing以及 翻译版）：

• Flink 将所有新生成的 sstable 备份到持久化存储（比如 HDFS，S3），并在新的 checkpoint 中引用。Flink 并不备份前一个 checkpoint 中已经存在的 sstable，而是在需要时仅引用他们，并记录下引用次数。Flink 还能够保证所有的 checkpoint 都不会引用已经删除的文件，因为 RocksDB 中文件删除是由压缩完成的，压缩后会将原来的内容合并写成一个新的 sstable。因此，Flink 增量 checkpoint 能够切断 checkpoint 历史。

以下转自Apache Flink 管理大型状态之增量 Checkpoint 详解，作者Stefan Ricther & Chris Ward，翻译 邱从贤（山智）


关于控制便捷性与性能之间平衡的策略可以参考此文档：

• Tuning Checkpoints and Large State

5.4.6 高级用法

• KeyedStateCheckpointOutputStream
• OperatorStateCheckpointOutputStream

5.4.7 性能对比

• Windowed Word Count Flink 不同 StateBackends 吞吐量对比
  
  此时使用 FileSystem 和 Memory 的吞吐差异不大，而使用 RocksDB 的吞吐仅其余两者的十分之一左右。
• Windowed Word Count Flink 不同 StateBackends 延迟对比
  
  可见FileSystem 和 Memory 作为 Backends 时，延迟基本一致且较低，而RocksDB在OnYarn模式下 QPS 2000以后延迟指数级上升

5.4.8 更多配置

可参考：

• Flink完整配置参数

以下配置在conf/flink-conf.yaml中配置，改变后需要重启Flink应用。

Key

Default

Type

Description

state.backend

(none)

String

保存和Checkpoint State的位置

state.backend.async

true

Boolean

StateBackend是否应在可能且可配置的情况下使用异步快照方法。 某StateBackend可能不支持/仅支持异步快照，会忽略此选项。

state.backend.fs.memory-threshold

1024

Integer

State数据文件大小的最小值。 所有大小小于该值的State Chunk都以内联方式存储在根Checkpoint元数据文件中。

state.backend.fs.write-buffer-size

4096

Intege

写入文件系统的Checkpoint流的写缓冲区的默认大小。 实际的写缓冲区大小为本选项和state.backend.fs.memory-threshold的最大值

state.backend.incremental

false

Boolean

StateBackend是否应创建增量Checkpoint（如果可能）。 对于增量Checkpoint，仅存储与前一个Checkpoint的差异，而不存储完整的Checkpoint状态。 某些StateBackend可能不支持，会忽略此选项

state.backend.local-recovery

false

Boolean

默认禁用。为此StateBackend配置本地恢复。 当前，本地恢复仅涵盖Keyed-StateBackend。 注意，目前MemoryStateBackend不支持本地恢复，请忽略此选项。

state.checkpoints.dir

(none)

String

用于在Flink支持的文件系统中存储Checkpoint的数据文件和元数据的默认目录。 必须从所有参与的进程/节点（即所有TaskManager和JobManager节点）访问存储路径

state.checkpoints.num-retained

1

Integer

要保留的最大已完成Checkpoint数

state.savepoints.dir

(none)

String

Savepoint的默认存储目录。 由StateBackend用于将Savepoint写入文件系统（适用于MemoryStateBackend，FsStateBackend，RocksDBStateBackend）

taskmanager.state.local.root-dirs

(none)

String

定义用于存储基于文件的State以进行本地恢复的根目录。 当前，本地恢复仅涵盖Keyed-StateBackend。 当前，MemoryStateBackend不支持本地恢复，请忽略此选项

5.5 巨大状态的调优

5.5.1 概述

为了使Flink应用程序在规模很大时仍能可靠运行，必须满足两个条件：

• 应用程序必须能够可靠地运行Checkpoint
• 发生故障后，程序重启后的资源必须足够赶上输入数据流

5.5.2 监控状态和Checkpoint

可参考Monitoring Checkpointing

Checkpoint关键指标：

• checkpoint_start_delay
  表示从Checkpoint被Checkpoint发起到算子真正开始执行Checkpoint时间 ，计算公式如下：
  checkpoint_start_delay = end_to_end_duration - synchronous_duration - asynchronous_duration

  如果该值持续高值，说明checkpoint barrier从Source流到算子的时间过长，一般就意味着出现了长期背压的情况。

• CheckpointBarrier对齐期间的数据缓存量
  在开启exactly-once语义后，多路input的算子收到一个input的Barrier就需要做对齐操作，在此期间就需要缓存数据。理想情况下缓存数据应该比较少，否则说明多个input的Bairrer长期不对齐。

5.5.3 Checkpoint调优

当Checkpoint耗时过长，意味着Checkpoint耗费太多资源，算子进展太少，虽然是异步Checkpoint，但依然会对整个应用性能产生影响。

此时就应该使用StreamExecutionEnvironment.getCheckpointConfig().setMinPauseBetweenCheckpoints调大Checkpoint之间最小间隔时间，该配置是定义了当前Checkpoint的结束与下一个Checkpoint的开始之间必须经过的最短时间间隔，官网图例如下：

• 图一为正常Checkpoint
• 图二为Checkpoint执行时间超过了设定的间隔，导致每次Checkpoint刚结束就开始下一个Checkpoint
• 图三表示采用了minPause限制，即使Checkpoint执行时间过久，也必须等到minPause后才能跑下一个Checkpoint

5.5.4 RocksDB调优

• 开启增量Checkpoint

• 将默认存在RocksDB中的Timer改为存在Java堆，谨慎使用！
  Timer较少时，可将state.backend.rocksdb.timer-service.factory设为heap，可获得更好的性能。但这样设置后，Timer状态就不能被异步快照存储了。而且可能增加Checkpoint时间，大小也不能超过能存限制**，谨慎使用！**

• Tuning RocksDB Memory

  1. 最直接的方式就是调大managed memory，可以显著提高性能。
     默认只有0.4比例，除非程序逻辑本身需要很多JavaHeap内存，否则可尽量多给managed memory

  2. 因为每个状态对应了一个RocksDB列族，每个列族都需要独有的write buffer，所以拥有很多状态的应用通常需要更多内存

  3. 可设state.backend.rocksdb.memory.managed：false来对比性能
     非托管模式中，除非使用了ColumnFamily，否则上线公式为140MB * num-states-across-all-tasks * num-slots（State包括Timer）

  4. 如果拥有很多状态的应用中观察到频繁地MemTable flushes，说明写入有瓶颈了，如果你不能给RocksDB更多内存了，那此时可以配置state.backend.rocksdb.memory.write-buffer-ratio将更多内存分配给write buffer。

  5. 专家可以尝试使用RocksDBOptionsFactory来调整arena block size， max background flush threads等，以减少MemTable flushes

         public class MyOptionsFactory implements ConfigurableRocksDBOptionsFactory {
     @Override
     public DBOptions createDBOptions(DBOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
         // increase the max background flush threads when we have many states in one operator,
         // which means we would have many column families in one DB instance.
         return currentOptions.setMaxBackgroundFlushes(4);
     }
     
     @Override
     public ColumnFamilyOptions createColumnOptions(
         ColumnFamilyOptions currentOptions, Collection<AutoCloseable> handlesToClose) {
         // decrease the arena block size from default 8MB to 1MB. 
         return currentOptions.setArenaBlockSize(1024 * 1024);
     }
     
     @Override
     public OptionsFactory configure(Configuration configuration) {
         return this;
     }
     }

5.5.5 资源规划

要想是的Flink Job运行可靠，一般可遵循以下步骤规划资源：

1. 正常运行应具有足够的容量，但应在恒定的背压下无法运行。
2. 在无故障且无背压运行程序所需的资源的基础上，还需提供一些额外的资源，以备在发生错误恢复时有足够资源来追上恢复任务期间累积的未处理数据。
   具体取决于任务恢复速度以及具体场景需要多快恢复
3. 短暂的背压是没问题的，他是负载高峰或要写入的外部系统突然变慢时的必要流控手段。
4. 一些算子（如巨大的window）会在每次触发时给下游算子带来负载高峰，所以下游算子并行度规划时需要考虑window大小、频率以及需要处理完成的速度
5. Job最大并行度应该设置足够，决定了使用savepoint来重设并行度时的上限

5.5.6 Checkpoint压缩

默认关闭。

目前只支持snappy compression algorithm (version 1.1.4):

ExecutionConfig executionConfig = new ExecutionConfig();
executionConfig.setUseSnapshotCompression(true);

注意，此选项不适用于增量快照，因为增量快照使用的是RocksDB的内部格式，始终使用开箱即用的snappy compression 。

Checkpoint压缩工作粒度为keyed state下的key-group，解压时也是此粒度，方便扩缩容。

5.5.7 Task本地恢复

5.5.7.1 现存问题-大状态任务恢复慢

Checkpoint时，每个Task都会生产一个状态快照，然后异步写入StateBackend。随后，每个Task会发送一个ACK给JM告知状态写入成功，该ACK是一个句柄，带有state在StateBackend中的位置。JM在收集完所有task的ACK后，将他们封装到一个Checkpoint对象中。

在恢复时，JM打开最近的Checkpoint，然后将其中的state文件句柄发回给各个对应的task以进行恢复。

使用分布式StateBackend的好处是容错性和方便扩缩容；坏处是必须通过网络远程读取访问，可能在大型状态场景导致很长的恢复时间，尽管可能只是因为很小的错误发生。

5.5.7.2 Task本地恢复

5.5.7.2.1 概述

核心思想就是每次Checkpoint时，写一份主副本到远程StateBackend，再写一份副本到本地（磁盘或内存）。

这样，在恢复时大多数task只需要找到之前本地state就能从本地恢复无需从远程下载。

5.5.7.2.2 两个副本间的关系

Checkpoint的StateBackend副本始终被认为是主副本，而本地副本是第二副本。而且他们格式不一定相同，比如存放在内存的本地状态就是Java对象。

Checkpoint时，主副本必须成功此次Checkpoint才能成功，而如果此时本地副本写入失败不会导致此次checkpoint失败。

主副本被JM确认和管理，而第二副本是由TM管理，生命周期可独立于主副本。比如保留3个最新主副本，而只保留一个本地最新副本。

恢复时，总是先尝试本地副本，如果失败就找远程主副本，如果还是失败就根据配置可找上一个checkpoint继续恢复。

更巧妙的是，即使Flink因为错误只写入部分状态到本地副本，也会尝试恢复这部分状态，然后再去远程副本恢复其他状态，因为远程状态总是完整的。

但如果TM挂了，则关联的本地状态也会全部丢失。

5.5.7.2.3 本地Checkpoint副本配置

默认关闭。

配置

默认

类型

说明

state.backend.local-recovery

false

Boolean

默认禁用。为此StateBackend配置本地恢复。
当前，本地恢复仅涵盖Keyed-StateBackend。
注意，目前MemoryStateBackend不支持本地恢复，请忽略此选项。

注意，目前非对齐Checkpoint不支持本地Checkpoint副本恢复。

5.5.7.2.4 不同StateBackend的本地Checkpoint副本

当前只支持KeyedStateBackend，未来会支持算子和timer的状态。

目前支持的StateBackend如下：

• FsStateBackend
  将状态复制到一个本地文件，会带来一定写入和磁盘空间开销。未来也许会做一个内存副本版本。
• RocksDBStateBackend
  • 增量Checkpoint
    根据RocksDB本地checkpoint机制。详解Details on task-local recovery for different state backends
  • 全量Checkpoint
    将状态复制到一个本地文件，会带来一定写入和磁盘空间开销。

5.5.7.2.5 保留分配的task调度

Task本地恢复模式中假定在错误后保留task调度，工作机制如下：

• 每个task记住之前的分配，请求同样的slot来开始恢复
• 如果该slot不可用，则task向RM请求一个全新的slot
  这样，即使一个TM不再可用，一个不能回归到之前位置的task不会再驱动其他正在恢复的任务从之前的slot中移出
• 这样做的好处是使得从本地恢复状态的task数量最大化，避免了夺取其他task之前的slot带来的级联效应（互相夺取slot）。

5.6 增量Checkpoint

5.6.1 概述

在减少Checkpoint花费的时间时，可首先考虑开启增量Checkpoint。 与完整Checkpoint相比，增量Checkpoint可以大大减少Checkpoint的时间，因为增量Checkpoint仅记录与先前完成的Checkpoint相比发生变化的部分，而不是生成完整数据。

一个增量checkpoint依赖之前的若干checkpoint构建。由于 RocksDB 内部存在 compaction 机制对 sst 文件进行合并，Flink 的增量快照也会定期重新设立起点（rebase），因此增量链条不会一直增长，旧快照包含的文件也会逐渐过期并被自动清理。

如果网络为瓶颈，则从增量Checkpoint恢复可能会较慢，因为需要读取更多delta增量文件；而如果CPU或IO为瓶颈，则从增量Checkpoint恢复更快，因为这种方式不会像全量或savepoint那样使用并解析Flink Key/Value 快照格式数据来重建本地RocksDB表。

增量Checkpoint开启方式：

new RocksDBStateBackend(path, TernaryBoolean.TRUE);

这里的Path是Checkpoint持久化目录，如hdfs://namenode:40010/flink/checkpoints 或file:///data/flink/checkpoints。
或使用配置state.backend.incremental: true开启

需要注意的是，只要开启了增量Checkpoint，则在WebUI上显示的Checkpointed Data Size就只代表增量Checkpoint数据大小，而不是整个State大小了。

5.6.2 原理

• 全量Checkpoint
  全量Checkpoint会在每个节点做备份数据时将数据遍历一遍，然后再写到外部存储中，这种情况会影响备份性能。下图ss是不可变的数据文件，manifest是元数据文件
  
• 增量Checkpoint
  此时只会将新产生的文件持久化，而不用管旧文件，提高Checkpoint性能。
  

5.6.3 调优

5.7 状态最佳实践

• Flink State 最佳实践
• 如何在 Flink 中规划 RocksDB 内存容量？

6 Task和Operator

6.1 Task

6.1.1 概述

是PhysicalGraph的节点，一个task是Job的基本组成单元，在Flink运行时被执行。

Task 精确封装了一个 Operator 或者 Operator Chain 的并行实例。

6.1.1 Sub-Task

一个Sub-Task是指处理数据流的一个partition的task。

该术语强调同一个Operator 或者 Operator Chain 拥有多个并行实例由多个Task执行。

6.2 Operator

6.2.1 概述

Operator即算子，是LogicalGraph的节点。

每个算子执行特定操作。

6.2.2 Operator与Task

分布式计算中，Flink 将算子（operator）的 subtask 按一定规则链接成若干task，每个 task 由TaskManager上的一个线程执行。

把算子链接成 task 能够有效减少线程间切换和缓冲的开销，在降低延迟的同时提高了整体吞吐量。

下图的 dataflow 由5个 subtasks 执行，因此具有5个并行线程运行：

• 上方那个Operator Chain是整个算子链的概要图
  可以看到source和map组成了一个chain在一个task执行；keyBy是一个单独的task；sink也是一个单独task
• 下方是加入了并行概念后的算子链视图
  因为source/map/keyBy 并行度都为2，所以共有4个Task；sink并行度为1；所以总共有5个task。

Operator Chain配置详情可参考：6.2.2

6.2.3 Operator Chain

6.2.3.1 概述

将两个连续的算子（即不会造成repatitioning的算子）连接即将他们放在同一个线程执行，同一个算子链内的算子直接交换数据，而不会进行序列化/反序列化或Flink网络相关操作，以获得更好的性能，优点：

• 减少线程切换
• 内部传输过程不会进行序列化/反序列化或Flink网络相关操作
• 减少数据在缓冲区的交换
• 减少了延迟的同时提高整体的吞吐量

Flink默认会尽可能将算子连接（比如两个map算组），但也可以自主控制：判断是否构成Chain的条件很苛刻，需要以下提交件全部都为true:

• 下游StreamNode的入边数为1，即只有一个input，而不是多个输入
• 上游StreamNode和下游StreamNode同在一个SlotSharingGroup
• 上游的ChainingStrategy不能是NEVER
  即只能是ALWAYS（可参与上下游连接，如FlatMap、Filter、Map）或HEAD（只能作为上游与下游连接，但不能与下游连接，如Source）
• 下游必须是ALWAYS
• 该StreamEdge的Partitioner必须是ForwardPartitioner，即发送和接收分区一一对应
• 该StreamEdge的ShuffleMode不能为BATCH，该模式下生产者先生产完整数据，然后消费者才开始消费数据；还有个模式是PIPELINED，此时生产者消费者同时完成生产消费；UNDEFINED的含义是由Flink推断到底采用以上哪种模式。
• 上下游并行度相同
• StreamGraph允许Chain机制。默认true，可通过pipeline.operator-chaining设置。

下面这幅图，展示了Source并行度为1，FlatMap、KeyAggregation、Sink并行度均为2，最终以5个并行的Task线程来执行:

可以看到，Key Agg和Sink因为符合Chain的条件被连接到一起，作为一个Task执行。

6.2.3.2 配置

可在job级别禁用 Operator Chain：

StreamExecutionEnvironment.disableOperatorChaining()

细粒度控制：

• 开启新的Operator Chain
  注意startNewChain必须用在算子之间，不能直接放在stream后面。

  以下例子filter是单独的task，但第一个map和第二个map组成了一个Operator Chain

  someStream.filter(...).map(...).startNewChain().map(...)

• 算子级别禁用Operator Chain

  someStream.map(...).disableChaining();

6.2.3.3 实现原理

以下为入度为1，出度为2的一个OperatorChain示例。

• OperatorChain为一个算子链的呈现类

• OperatorChain对外部来说就是一个整体，只需要将input数据传输给该算子链的HeadOperator即可
  比如上图的算子链就可以看作是入度为1，出度为2的一个算子整体

• 上图中的实线就对应了JobEdge

• 上图中的虚线就是OperatorChain内部数据传输ChainingOutput，不会经过序列化/反序列化、网络传输，而是直接通过方法传递处理。

  OperatorChain.ChainingOutput如下

  static class ChainingOutput<T> implements WatermarkGaugeExposingOutput<StreamRecord<T>> {
      // operatorChain的下游算子
      protected final OneInputStreamOperator<T, ?> operator;
      protected final Counter numRecordsIn;
      protected final WatermarkGauge watermarkGauge = new WatermarkGauge();
  protected final StreamStatusProvider streamStatusProvider;
  
  // 为一个算子标记side output
  @Nullable
  protected final OutputTag&lt;T&gt; outputTag;
  
  public ChainingOutput(
          OneInputStreamOperator&lt;T, ?&gt; operator,
          StreamStatusProvider streamStatusProvider,
          @Nullable OutputTag&lt;T&gt; outputTag) {
      // operatorChain的下游算子   
      this.operator = operator;
      ...
      this.streamStatusProvider = streamStatusProvider;
      this.outputTag = outputTag;
  }
  
  @Override
  public void collect(StreamRecord&lt;T&gt; record) {
      if (this.outputTag != null) {
          // we are not responsible for emitting to the main output.
          return;
      }
  
      pushToOperator(record);
  }
  ...
  protected &lt;X&gt; void pushToOperator(StreamRecord&lt;X&gt; record) {
      try {
          // we know that the given outputTag matches our OutputTag so the record
          // must be of the type that our operator expects.
          @SuppressWarnings("unchecked")
          StreamRecord&lt;T&gt; castRecord = (StreamRecord&lt;T&gt;) record;
  
          numRecordsIn.inc();
          operator.setKeyContextElement1(castRecord);
          // 直接调用下游算子处理数据
          operator.processElement(castRecord);
      }
      catch (Exception e) {
          throw new ExceptionInChainedOperatorException(e);
      }
  }
  ...
  }

6.2.4 Flink 自带 Operator

首先，Source和Sink是特殊的算子，用来数据摄取和输出。

6.2.5 Operator 并行度

Operator 并行度配置可参考这里。

6.2.6 Operator数据交换与分区策略

DataStream可以在两个算子间传输数据，有以下两种模式：

• 一对一
  例如上图中Source 和 map()算子之间。

  可保留元素的分区和排序信息（也就是说map()算子的1号实例可以相同顺序看到跟Source算子的1号实例生产顺序相同的元素）。

• 重分发-类似MR Shuffle
  例如上图中的 map()和 keyBy/window算子 之间，以及 keyBy/window和 Sink 之间。

  会更改数据所在Stream分区。注意此时只能保证一个算子subtask发到一个下游算子subtask的元素顺序性。如上图keyBy/window 的 subtask[2] 接收到的 map() 的 subtask[1]的数据有序，但发送到Sink的所有数据中，无法确定不同key的聚合结果的到达顺序。

  每个算子subtask发送数据到不同的下游算子subtask，分发依据是具体的transformation(相关方法在org.apache.flink.streaming.api.datastream.DataStream)：

  • keyBy
    按照key的值hash后重分区到某个下游算子实例

  • broadcast
    广播到所有下游算子实例分区

  • rebalance
    轮询分配到下游算子实例分区

  • global
    全部分配到第一个下游算子实例分区

  • shuffle
    随机均匀分配到下游算子实例分区

  • forward
    上下游并行度一致时，发送到对应的位于本地的下游算子分区

  • rescale
    轮询方式将输出的元素均匀分发到下游分区的子集。

    子集构建依赖于上游和下游算子的并行度。

    • 比如上游算子并行度2，下游为4，此时每个上游算子轮询各自分发到下游的两个算子。
    • 如果上游并行度4，下游为2，此时每两个上游算子分发到一个下游算子。
    • 如果不是倍数，则下游分发的源头数目不一致

6.3 TaskSlot 和 资源

6.3.1 概述

每个TaskManager都是一个 JVM 进程，并且可以在不同的Slot线程中执行多个 subtasks，Slot控制 TaskManager 并发执行 task 的数量，每个TaskManager至少一个Slot。他们的关系示例如下图，此时作业基本并行度为2，一个taskManager的slot数量为3：

6.3.2 Slot与资源隔离

每个 Slot是TaskManager资源分配和调度最小单元，代表一份固定资源子集。例如，具有三个 slot 的 TaskManager 会将其管理的内存资源平均分成三等份给每个 slot。 划分资源意味着 subtask 之间不会竞争资源，但这也意味着它们只能拥有固定的划分了的资源。

注意Slot并没有实现 CPU 隔离，当前 Slot 之间只实现了内存资源隔离划分。

通过调整 slot 的数量，用户可以决定 subtask 的隔离方式:

• 每个 TaskManager 只有一个 slot
  意味着每个 task 在一个单独的 JVM 中运行（即每个task独享一个TM进程） 。但可以注意到，目前Flink1.11版本已经支持多个Task线程共享一个Slot，所以以上结论已经不再适用。

• 每个 TaskManager 多个 slot
  意味着多个 subtask 共享同一个 JVM，分别在各自slot线程运行。

  在同一个 JVM 中运行的Task会共享 TCP 连接（通过IO多路复用）和心跳信息，可以减少数据的网络传输。它们还可能共享数据集和数据结构，从而降低每个 task 的性能开销。

6.3.3 Subtask共享slot

默认情况下，Flink 允许 subtask们 共享 slot，即使它们是不同 task 的 subtasks，只要它们来自同一个 job就行。因此，一个 slot 在极端场景下甚至可能会负责这个 job 的整个执行pipeline！

允许 slot sharing 有两个好处：

• Flink 集群的Slot数量总数需要与 job 中使用的最高并行度(highest parallelism)完全相同。这样不需要计算作业总共包含多少个 tasks（具有不同并行度）。

• 更好的资源利用率
  如果不能共享slot ，则简单的 subtask（比如source/ map等）将会占用和复杂的 subtask （如window）一样多的资源。

  而通过slot共享，将之前示例中的job最大并行度从 2 增加到 6 就可以完全利用 按slot分隔 的资源，同时确保开销大的 subtask 在 所有TaskManager上均匀分布：
  

6.3.4 slot与线程

一个Slot并不一定只有一个线程运行，比如上图中一个Task Slot内部就运行了2-3个线程！

这些线程共同分享Slot获得的分隔内存资源。

不同算子是否能运行在一个Slot取决于SlotSharingGroup

6.3.5 slot数量设置

根据经验，合理的 slots 数量应该和 CPU 核数相同。在使用超线程（hyper-threading）时，每个 slot 将会占用 2 个或更多的硬件线程上下文（hardware thread contexts）。

6.3.6 SlotSharingGroup

6.3.6.1 概述

Flink会将相同SlotSharingGroup的算子放到相同的slot执行，而将SlotSharingGroup不同的算子放到其他slot，这可用来隔离slot。

如果所有input算子都在相同SlotSharingGroup，则下游算子的SlotSharingGroup继承自input算子。

请记住，SlotSharingGroup软定义了不同的task(JobVertex)是否可在一个Slot中运行

6.3.6.2 配置

默认的slot sharing group的名字是default，算子也可以显示地放入指定名字的slot sharing group：

someStream.filter(...).slotSharingGroup("name");

6.3.7 CoLocationGroup

CoLocationGroup是JobVertex组，硬性定义了某个JobVertex的第i个子任务必须和其他所有JobVertex的第i个子任务运行在相同TaskManager上。

6.4 Task运行模型

6.4.1 概述

• Task运行在一个独占线程中，多个Task线程可以共享一份Slot资源
• Source Task使用SourceFunction生产数据
• 运算Task负责从InputGate读取数据，调用OperatorChain，最终结果输出到ResultSubPartition，通过Netty发送给下游算子的InputChannel。
• 最后是Sink的InputChannel接收到数据后，通过调用SinkFunction相关方法发送到目的地。

6.4.2 概述

7 容错性和高可用

7.1 概述

当Flink某个任务失败挂掉时，会对失败任务和其他受影响任务重启来恢复整个作业。

Flink中主要有两类策略来控制任务重启：

• 重启策略
  决定了是否重启、何时重启失败和受影响任务
• Failover策略
  决定了job中的哪些任务应该被重启。

可参考

• 任务失败恢复

如果Flink部署在yan上，则需要依赖Yarn来实现高可用，Yarn会对失败挂掉的JobManager(AM)重启，最大重试次数的配置是yarn-site.xml的yarn.resourcemanager.am.max-attempts。

另一方面，Flink还支持高可用配置。

7.2 重启策略

优先级最高的是job自己指定的，然后才是flink-conf.xml中指定的：

采用restart-strategy配置。

7.2.1 不允许Checkpoint时默认采用none

当不允许checkpoint时，采用的是none(也可以填off、disable)策略，即不重启

可用配置项：

• restart-strategy: none

代码：

val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.noRestart())

7.2.2 开启Checkpoint时默认fixeddelay

开启checkpoint时，不配置重启策略时默认采用fixeddelay(也可以填fixed-delay、failure-rate)，是一种固定间隔重启的策略，默认会1秒间隔来进行Integer.MAX_VALUE次重启尝试，超过最大次数就会导致job最终失败了。

可用配置项：

• restart-strategy
  默认fixed-delay
• restart-strategy.fixed-delay.attempts
  默认1，即当job失败时重试一次
• restart-strategy.fixed-delay.delay
  重试间隔，默认1 s

代码设置：

val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
  3, // number of restart attempts
  Time.of(10, TimeUnit.SECONDS) // delay
))

7.2.3 开启Checkpoint时还可选failurerate

开启checkpoint时 还可选failurerate, (也可以填failure-rate）。按失败率重启。

失败率是指每个时间间隔内发生的失败次数。

当失败率超过设定阈值，则job最终失败了。

可用配置项：

• restart-strategy
  failure-rate
• restart-strategy.failure-rate.delay
  默认 1 s，指定重启间隔时间，比如20 s
• restart-strategy.failure-rate.failure-rate-interval
  默认1 min，指定计算失败率的时间间隔
• restart-strategy.failure-rate.max-failures-per-interval
  默认 1， 失败率最大值

代码设置：

val env = ExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.failureRateRestart(
  3, // 每个时间间隔的最大故障次数
  Time.of(5, TimeUnit.MINUTES), // 测量故障率的时间间隔
  Time.of(10, TimeUnit.SECONDS) // 延时
))

7.3 Failover策略

配置项为jobmanager.execution.failover-strategy，Failover策略 相关接口为FailoverStrategy

• 可选值full，重启job所有task。
  RestartAllFailoverStrategy

• 默认值region，当task失败时，重启所有可能被该出错task影响的所有task。
  RestartPipelinedRegionFailoverStrategy

  该策略将task分组为不同region，当任务失败时就计算必须重启以恢复job的最小范围的region set。

  一个Region是指一个pipeline，是一些task的集合，这些task通过pipeline交换数据通讯：

  • DataStream 和 流式 Table/SQL job的所有数据交换都是 Pipelined 形式的。
  • 批处理式 Table/SQL job的所有数据交换默认都是 Batch 形式的。
  • DataSet job中的数据交换形式会根据 ExecutionConfig 中配置的 ExecutionMode决定。

  需要重启的 Region 的判断逻辑如下：

  • 出错 Task 所在 Region 需要重启。
  • 如果要重启的 Region 需要消费的数据有部分无法访问（丢失或损坏），产出该部分数据的 Region 也需要重启。
  • 需要重启的 Region 的下游 Region 也需要重启。这是出于保障数据一致性的考虑，因为一些非确定性的计算或者记录分发可导致同一个 ResultPartition 每次产生时包含的数据都不相同。

• 单点重启
  可能无法保证一致性，但资源开销最小。

7.4 TaskManager和JobManager高可用

可参考

• JobManager High Availability (HA)
• FlinkWiki-JobManager High Availability
• Apache Flink 官方文档–作业管理器(JM, JobManager)高可用（HA）

7.4.1 TaskManager HA

由JM保证TM恢复期间的跨TM一致性。

7.4.2 JobManager单点问题与HA

因为JM用来调度任务、管理资源，所以如果他挂了就导致整个程序失败了，这就是JM单点问题。

我们可以配置JM HA高可用，即当JM失败后进行恢复，适用于 Standalone和Yarn Cluster模式，这里我们主要分析Yarn Cluster模式。

7.4.3 Yarn Cluster模式 JM ZK HA

7.4.3.1 ZK HA概述

相关源码在ZooKeeperHaServices。

ZK存储的Flink相关数据结构如下：

• flink
  ZK存放Flink数据的根目录，通过high-availability.zookeeper.path.root配置
• cluster_id
  某个Flink集群的数据根节点。包括Standalone模式和Flink集群模式的Cluster
• job-id
  Flink集群上运行的job信息，包括Checkpoint信息
• checkpoints
  触发Checkpoint快照后，每个TM会生成Snapshot，并把对应的句柄传递给JM，JM会创建全局Checkpoint并持久化，最后将文件句柄写入ZK。
• persisted_job_graph
  JM接收到客户端提交的JobGraph后会生成并持久化ExecutionGraph。具体来说，JM会先将ExecutionGraph持久化到ZK，再开始调度Task到TM执行。

7.4.3.2 配置yarn-site.xml

配置yarn-site.xml内的yarn.resourcemanager.am.max-attempts

即am重试次数，要算上初次启动，默认值为2，即当设置了JM HA后允许JM出错重启一次

7.4.3.3 配置 flink-conf.yaml ZK

要保证JM服务自身在恢复期间的一致性，就必须使用第三方服务来存储少量的恢复用的元数据（比如最后提交的Checkpoint等），以及帮助选举哪个备选JM当Leader，避免脑裂。

• high-availability
  定义了HA模式，可选项如下。

  • 默认NONE，不启用HA
  • ZOOKEEPER，ZK模式的HA
  • 恢复工厂类的全限定名的HA

• high-availability.cluster-id
  定义了Flink集群ID，用来隔离不同Flink集群。

  需要为Standalone模式设置，会在YARN和Mesos中自动推断。

• high-availability.storageDir
  Flink持久化元数据的文件系统路径(URL)

• high-availability.zookeeper.path.root
  Flink存放集群节点元数据信息的ZK上的根目录路径，ZK上存了指向该数据的指针信息。

• high-availability.zookeeper.quorum
  Flink HA模式时使用的ZK法定节点数(quorum)

• high-availability.jobmanager.port

更多Flink HA Zk高级配置可参考Advanced High-availability ZooKeeper Options

7.4.3.4 配置flink-conf.yaml AM/JM重启

配置flink-conf.yaml内的yarn.application-attempts和yarn.application-attempt-failures-validity-interval。

Flink On Yarn时需要依赖Yarn来实现高可用，Yarn会对失败挂掉的JobManager(AM)重启，最大重试次数的配置是yarn-site.xml的yarn.resourcemanager.am.max-attempts。

Flink的yarn-client有一些配置可以控制在container失败的情况下的行为，也可通过$FLINK_HOME/conf/flink-conf.yaml或启动yarn-session.sh时以-D参数指定：

• yarn.application-attempts
  ApplicationMaster（运行着JobManager）及其TaskManager Container的最大失败重试次数。

  当没有设置HA时默认值为1，此时若AM挂掉就直接导致整个flink yarn session失败了。

  如果设为较高的值，使得可在失败时Yarn可多次尝试重启AM，但会导致整个Flink集群重启，而Yarn Client会丢失到该Flink Cluster的连接，JobManager的地址也会改变，所以需要在重启后手动设置JobManager的 host:port，所以推荐保留该值为1。

  如果超过该阈值，不会再重启AM，该yarn session上提交的任务也会全部停止。比如设为5时，代表初始化1次、最大重试4次。

  如果Yarn有一些额外操作（如抢占、硬件故障或reboot、NM resync等）需要重启Application，则不会计算在次数内，参考Jian Fang’s blog

  注意该值不能超过yarn中yarn.resourcemanager.am.max-attempts配置的最大重启次数。

• yarn.application-attempt-failures-validity-interval
  默认为10000，单位毫秒

  定义yarn.application-attempts统计的时间窗口。如果设为-1则为全局统计。

  也就是说，如果定义的时间窗口内统计到的attempts达到了阈值，则会停止尝试重启，导致Job失败！

  注意Yarn Container在不同版本中的行为不太相同，在大于等于2.6.0后应将yarn.application-attempt-failures-validity-interval设置为Flink的Akka超时值。

  相关连接：

  • PR
  • 实例

7.4.3.5 Flink HA ZK Mode 配置实例

flink-conf.yaml:

high-availability: zookeeper
high-availability.zookeeper.quorum: localhost:2181
high-availability.storageDir: hdfs:///flink/recovery
high-availability.zookeeper.path.root: /flink
yarn.application-attempts: 10

8 BackPressure背压/反压

8.1 概述

如果在Flink WebUI上看到某个task发生back pressure warning，那么通俗地说，这就意味着数据生产速度大于下游算子消费速度，俗称反压/背压。

数据在Flink中从Source到Sink向下流动，而反压是反向传播的。

就拿最简单的Source -> Sink来说，如果观察到Source出现反压，则说明Sink消费速度已经跟不上Source生产速度了，所以向上游Source算子产生反压。

8.2 反压指标采样

Flink JM会周期性地调用Task.isBackPressured()方法，以从运行中的task中采样，监控反压指标。

默认每次采样会为每个task每50ms采样100次，可在WebUI观察该指标（60秒刷新一次，避免TM过载），比如0.01表示百分之1的样本发生反压。

该指标有几种情况：

• OK: 0 <= Ratio <= 0.10

• LOW: 0.10 < Ratio <= 0.5

• HIGH: 0.5 < Ratio <= 1

  @Override
  public boolean isBackPressured() {
  if (invokable == null || consumableNotifyingPartitionWriters.length == 0 || !isRunning()) {
  return false;
  }
  final CompletableFuture[] outputFutures = new CompletableFuture[consumableNotifyingPartitionWriters.length];
  for (int i = 0; i < outputFutures.length; ++i) {
  outputFutures[i] = consumableNotifyingPartitionWriters[i].getAvailableFuture();
  }
  return !CompletableFuture.allOf(outputFutures).isDone();
  }

目前判断是否产生背压是通过output buffer可用性，如果没有足够的buffer可用于输出说明该Taskb受到了来自下游的反压。

8.3 JM配置

• web.backpressure.refresh-interval
  有效的反压结果被废弃并重新进行采样的时间间隔 (默认: 60000ms, 即1 min)。
• web.backpressure.num-samples
  用于确定反压采样的样本数 (默认: 100)。
• web.backpressure.delay-between-samples
  用于确定反压采样的间隔时间 (默认: 50 (ms))。

总结

本章主要介绍了Flink的一些基本概念，下一章讲Flink的下载、安装和启动，请点击Flink系列2-安装和启动

好文推荐

• Apache Flink
• Flink在唯品会的实践应用
• 大数据处理引擎 Apache Flink 全梳理
• Flink架构、原理与部署测试
• Apache Flink - 架构和拓扑
• Flink原理及架构
• Flink-StreamWindow

参考文档

• Apache Flink
• 《深入理解Flink-实时大数据处理时间》
• Flink的水印机制到底是怎么回事
• Flink内部通信组件Netty和Akka的区别
• 为什么 spark 2.0 底层通信不用 Akka 而转用 netty ？
• Flink的迁移之路，2年处理效果提升5倍
• Flink 原理与实现：理解 Flink 中的计算资源
• Flink架构分析之资源分配
• 《Flink Runtime Architecture》
  • 作者：朱翥
  • 出处：阿里巴巴-Flink极客训练营
• Apache Flink 零基础入门（一&二）：基础概念解析
  • 作者：陈守元、戴资力
  • 出处：Ververica
• Apache Flink 进阶教程（一）：Runtime 核心机制剖析
  • 作者：高赟
  • 出处：Ververica
• Apache Flink 零基础入门（七）：状态管理及容错机制
  • 作者：孙梦瑶
  • 出处：Ververica
• Flink 从checkpoint恢复，并行度改变后，状态重分配
  • 作者：lvwenyuan_1
• 日均万亿条数据如何处理？爱奇艺实时计算平台这样做
  • 作者：梁建煌
• 大数据框架：Spark 生态实时流计算
  • 作者：加米谷大数据
• Structured Streaming VS Flink
  • 作者：浪尖
• Apache Flink 进阶教程（三）：Checkpoint 的应用实践
  • 作者：唐云(茶干)
  • 出处：Ververica
• Apache Flink状态管理和容错机制介绍
  • 作者：施晓罡
  • 出处：Ververica
• 流计算框架 Flink 与 Storm 的性能对比
  • 作者：孙梦瑶
  • 出处：Ververica
• 谈谈流计算中的『Exactly Once』特性
  • 作者：周凯波（宝牛）
  • 出处：Ververica
• Flink状态管理详解：Keyed State和Operator List State深度解析
  • 作者：大数据从业者
  • 出处：cnblogs
• 一文搞懂 Flink 的 Exactly Once 和 At Least Once
  • 作者：范瑞
  • 出处：Ververica
• 如何在 Flink 中规划 RocksDB 内存容量？
  • 作者：Stefan Richter
  • 翻译：毛家琦
  • 校对：胡争
  • 出处：Ververica
• Apache Flink 管理大型状态之增量 Checkpoint 详解
  • 作者：Stefan Ricther & Chris Ward
  • 翻译：邱从贤（山智）