flink状态与容错

容错:在服务器出现问题的情况下，还可以保证数据不丢失(在出错误的情况下，还能保证一个数据只被处理一次)

checkpoint

为了保证flink的容错，需要为状态添加checkpoint(检查点)。checkpoint使得flink能够恢复状态和在流中的位置，从而向应用提供和无故障执行时一样的语义。

前提条件

Flink的checkpoint机制会和持久化存储进行交互，读写流与状态。一般需要：

一个能够回放一段时间内数据的持久化数据源，例如持久化消息队列(apache kafka,RabbitMQ,Amazon Kinesis,Google PubSub等)或文件系统(HDFS等)
存放状态的持久化存储，通常为分布式文件系统(比如HDFS等)

开启与配置Checkpoint

默认情况下 checkpoint 是禁用的。通过调用 StreamExecutionEnvironment 的 enableCheckpointing(n) 来启用 checkpoint，里面的 n 是进行 checkpoint 的间隔，单位毫秒。

Checkpoint 其他的属性包括：

精确一次（exactly-once）对比至少一次（at-least-once）：你可以选择向 enableCheckpointing(long interval, CheckpointingMode mode) 方法中传入一个模式来选择使用两种保证等级中的哪一种。对于大多数应用来说，精确一次是较好的选择。至少一次可能与某些延迟超低（始终只有几毫秒）的应用的关联较大。
checkpoint 超时：如果 checkpoint 执行的时间超过了该配置的阈值，还在进行中的 checkpoint 操作就会被抛弃。
checkpoints 之间的最小时间：该属性定义在 checkpoint 之间需要多久的时间，以确保流应用在 checkpoint 之间有足够的进展。如果值设置为了 5000，无论 checkpoint 持续时间与间隔是多久，在前一个 checkpoint 完成时的至少五秒后会才开始下一个 checkpoint。

往往使用“checkpoints 之间的最小时间”来配置应用会比 checkpoint 间隔容易很多，因为“checkpoints 之间的最小时间”在 checkpoint 的执行时间超过平均值时不会受到影响（例如如果目标的存储系统忽然变得很慢）。

注意这个值也意味着并发 checkpoint 的数目是一。
checkpoint 可容忍连续失败次数：该属性定义可容忍多少次连续的 checkpoint 失败。超过这个阈值之后会触发作业错误 fail over。默认次数为“0”，这意味着不容忍 checkpoint 失败，作业将在第一次 checkpoint 失败时fail over。
并发 checkpoint 的数目: 默认情况下，在上一个 checkpoint 未完成（失败或者成功）的情况下，系统不会触发另一个 checkpoint。这确保了拓扑不会在 checkpoint 上花费太多时间，从而影响正常的处理流程。不过允许多个 checkpoint 并行进行是可行的，对于有确定的处理延迟（例如某方法所调用比较耗时的外部服务），但是仍然想进行频繁的 checkpoint 去最小化故障后重跑的 pipelines 来说，是有意义的。

该选项不能和 “checkpoints 间的最小时间"同时使用。
官网地址

https://nightlies.apache.org/flink/flink-docs-release-1.15/zh/docs/dev/datastream/fault-tolerance/checkpointing/

在代码中开启

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 每 1000ms 开始一次 checkpoint
env.enableCheckpointing(1000);

// 高级选项：

// 设置模式为精确一次 (这是默认值)
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 确认 checkpoints 之间的时间会进行 500 ms
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

// Checkpoint 必须在一分钟内完成，否则就会被抛弃
env.getCheckpointConfig().setCheckpointTimeout(60000);

// 允许两个连续的 checkpoint 错误
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
        
// 同一时间只允许一个 checkpoint 进行
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
env.getCheckpointConfig().setExternalizedCheckpointCleanup(
        ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

// 开启实验性的 unaligned checkpoints
env.getCheckpointConfig().enableUnalignedCheckpoints();

在配置文件中开启

vim flink-conf.yaml

execution.checkpointing.interval: 5000
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.min-pause: 0
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.timeout: 10min
execution.checkpointing.tolerable-failed-checkpoints: 0
execution.checkpointing.unaligned: false
state.backend: hashmap
state.checkpoints.dir: hdfs://master:9000/flink/checkpoint

使用Checkpoint

第一次提交任务之间提交

# -s 指定恢复任务的位置
flink run -t yarn-session -p 3 -Dyarn.application.id=application_1717039073374_0009  -c com.shujia.flink.state.Demo5ExactlyOnceSInkKafka -s hdfs://master:9000/flink/checkpoint/aa5c16e40767a315674780ba01a92fb3/chk-2 flink-1.0.jar

重启任务时基于hdfs中的快照重启

# -s 指定恢复任务的位置
flink run -t yarn-session -p 3 -Dyarn.application.id=application_1717039073374_0009  -c com.shujia.flink.state.Demo5ExactlyOnceSInkKafka -s hdfs://master:9000/flink/checkpoint/aa5c16e40767a315674780ba01a92fb3/chk-2 flink-1.0.jar

开启持久化的checkpoint存储在HDFS中

基于上一次提交的任务的快照启动任务

向端口打进一条java

数据没有丢失

Checkpoint原理

但是整个过程是很抽象的

从图中可以看出，只有需要进行计算的算子才需要状态(2.01KB)

有状态算子

把之前的结果看作是状态，用之前的结果进行计算是有状态计算

 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> wordsDS = env.socketTextStream("master", 8888);

        //分组
        KeyedStream<String, String> keyByDS = wordsDS.keyBy(word -> word);

        /*
         * process算子是flink提供的一个底层算子，可以获取到flink底层的状态，时间和数据
         */
        DataStream<Tuple2<String, Integer>> countWords = keyByDS
                .process(new KeyedProcessFunction<String, String, Tuple2<String, Integer>>() {
                    /**
                     * 保存之前的结果(状态)
                     * 问题：统一个task中的数据共享统一个count变量
                     * int count = 0;
                     * 需要为每一个key保存一个结果
                     * 使用单词作为key，数量作为value
                     * 问题：需要hashmap保存之前计算的中间结果,flink的checkpoint不会将hashmap中的数据持久化到hdfs中
                     * 所有任务失败重启会丢失之前的结果
                     */
                    int count = 0;

                    @Override
                    public void processElement(String word,
                                               KeyedProcessFunction<String, String,
                                                       Tuple2<String, Integer>>.Context context,
                                               Collector<Tuple2<String, Integer>> out) throws Exception {

                        /**
                         * processElement方法每一条数据执行一次
                         * word 一行数据
                         * context 上下文对象
                         * out 用于将处理结果发送到下游
                         */
                        count++;

                        out.collect(Tuple2.of(word, count));


                    }
                });


        countWords.print();
        env.execute();

问题：同一个task中的数据是共享同一个count变量，所以不可以实现累加计算
修改：需要为每一个key保存一个结果，使用hashMap为每一个key做存储

  //为每一位key保存一个结果
                    //使用单词作为key,数量作为value
                    HashMap<String,Integer> map = new HashMap<>();

                    @Override
                    public void processElement(String word,
                                               KeyedProcessFunction<String, String,
                                                       Tuple2<String, Integer>>.Context context,
                                               Collector<Tuple2<String, Integer>> out) throws Exception {

                        Integer count = map.getOrDefault(word, 0);
                        /**
                         * processElement方法每一条数据执行一次
                         * word 一行数据
                         * context 上下文对象
                         * out 用于将处理结果发送到下游
                         */
//                        System.out.println(map);
                        count++;

                        out.collect(Tuple2.of(word, count));
                        //更新之前的结果
                        map.put(word,count);

可以做有状态的计算，但是不具备容错的特点
如果运行过程中发生了故障，再次重新启动后数据丢失。在flink持续计算中丢失数据显然代价是比较大的

如果我们去提交代码，发现checkpoint不会做持久化，原因是HashMap是Java中的算子，不能做持久化

如果我们想要做持久化，就要用到flink中的KeyBy State

使用 Keyed State

keyed state 接口提供不同类型状态的访问接口，这些状态都作用于当前输入数据的 key 下。换句话说，这些状态仅可在 KeyedStream 上使用，在Java/Scala API上可以通过 stream.keyBy(...) 得到 KeyedStream，在Python API上可以通过 stream.key_by(...) 得到 KeyedStream。

接下来，我们会介绍不同类型的状态，然后介绍如何使用他们。所有支持的状态类型如下所示：

ValueState<T>: 保存一个可以更新和检索的值（如上所述，每个值都对应到当前的输入数据的 key，因此算子接收到的每个 key 都可能对应一个值）。这个值可以通过 update(T) 进行更新，通过 T value() 进行检索。
ListState<T>: 保存一个元素的列表。可以往这个列表中追加数据，并在当前的列表上进行检索。可以通过 add(T) 或者 addAll(List<T>) 进行添加元素，通过 Iterable<T> get() 获得整个列表。还可以通过 update(List<T>) 覆盖当前的列表。
ReducingState<T>: 保存一个单值，表示添加到状态的所有值的聚合。接口与 ListState 类似，但使用 add(T) 增加元素，会使用提供的 ReduceFunction 进行聚合。
AggregatingState<IN, OUT>: 保留一个单值，表示添加到状态的所有值的聚合。和 ReducingState 相反的是, 聚合类型可能与添加到状态的元素的类型不同。接口与 ListState 类似，但使用 add(IN) 添加的元素会用指定的 AggregateFunction 进行聚合。
MapState<UK, UV>: 维护了一个映射列表。你可以添加键值对到状态中，也可以获得反映当前所有映射的迭代器。使用 put(UK，UV) 或者 putAll(Map<UK，UV>) 添加映射。使用 get(UK) 检索特定 key。使用 entries()，keys() 和 values() 分别检索映射、键和值的可迭代视图。你还可以通过 isEmpty() 来判断是否包含任何键值对。

所有类型的状态还有一个clear() 方法，清除当前 key 下的状态数据，也就是当前输入元素的 key。

请牢记，这些状态对象仅用于与状态交互。状态本身不一定存储在内存中，还可能在磁盘或其他位置。另外需要牢记的是从状态中获取的值取决于输入元素所代表的 key。因此，在不同 key 上调用同一个接口，可能得到不同的值。

你必须创建一个 StateDescriptor，才能得到对应的状态句柄。这保存了状态名称（正如我们稍后将看到的，你可以创建多个状态，并且它们必须具有唯一的名称以便可以引用它们），状态所持有值的类型，并且可能包含用户指定的函数，例如ReduceFunction。根据不同的状态类型，可以创建ValueStateDescriptor，ListStateDescriptor， AggregatingStateDescriptor, ReducingStateDescriptor 或 MapStateDescriptor。

状态通过 RuntimeContext 进行访问，因此只能在 rich functions 中使用。请参阅这里获取相关信息，但是我们很快也会看到一个例子。RichFunction 中 RuntimeContext 提供如下方法：

ValueState<T> getState(ValueStateDescriptor<T>)
ReducingState<T> getReducingState(ReducingStateDescriptor<T>)
ListState<T> getListState(ListStateDescriptor<T>)
AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT>)
MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV>)

官网：https://nightlies.apache.org/flink/flink-docs-release-1.15/zh/docs/dev/datastream/fault-tolerance/state/#使用-keyed-state

唯一一次的概念

一致性级别

1.至少一次（At Least Once）= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

2.最多一次（At Most Once）= ACK级别设置为0

3.对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。

    优缺点：At Least Once可以保证数据不丢失，但是不能保证数据不重复；
    
           At Most Once可以保证数据不重复，但是不能保证数据不丢失。

           Exactly Once--幂等性和事务可以保障数据精确一次

1.数据生产端

1.保证数据不重复

幂等性：kafka 0.11之后，Producer的send操作现在都是幂等的，在任何导致producer重试的情况下，相同的消息，如果被producer发送多次，也只会被写入Kafka一次
事务的支持 -- 保证数据不重复

1.保证数据不丢失

ACKS机制+副本

为避免生产者生产数据到节点的过程中服务器崩溃而造成数据丢失，引入了ACKS机制

acks机制：

acks=1(默认):当主节点写入成功后，就会返回成功，如果这时候主节点所在的节点挂了，刚写入的数据会丢失

acks=0:生产者只负责生产数据，不负责验证数据是否写入成功，会丢失数据，写入的性能好

acks=-1/all:生产者生产数据后必须等待所有副本都同步成功才会返回成功，不会丢失数据。性能差

2.数据消费端

Flink分布式快照保存数计算的状态和消费的偏移量，保证程序重启之后不丢失状态和消费偏移量

有状态的流处理，内部每个算子任务都可以有自己的状态。对于流处理器内部来说，所谓的状态一致性，其实就是我们所说的计算结果要保证准确。一条数据都不应该丢失，也不应该重复计算。在遇到故障时可以恢复状态，恢复以后的重新计算，结果应该也是完全正确的。

2.消费端为了保证数据不丢失

数据是一条流式的，可能在处理的任何时间点中失败，恢复的时候，需要找到上一个checkpoint记录的时间点，重新再读一下。这样才能保证数据的不丢失

checkpoint会定时将flink计算的状态保存到hdfs

如果在checkpoint持久化中间过程中节点崩溃，如果使用的sock...源，那么丢失的部分就丢失了。原因是sock源只能读一次，不支持重复读。所有也就是为什么要将数据源要改成Kafka

3.数据发送端

假如checkpoint做持久化的间隔是5秒。在发送数据的时候，如果在中间过程第7秒节点崩溃，那么返回上一个时间节点第五秒的时候，前面的数据已经存进hdfs中了。那么回到第五秒后再去读数据，第7秒的数据也会被读进去，数据就发生了重复。

3.Sink端为了保证数据的唯一一次

1.聚合计算

如果任务再执行的过程中失败了，可以恢复到上一次成功的checkpoint的位置，保证计算状态和偏移量不丢失，保证数据处理唯一一次

2.非聚合计算

需要将两次checkpoint中的数据放到一个事务中，上一次checkpoint完成时开启事务。下一次checkpoint完成时提交事务，会增加数据处理的延迟。

标签：状态,flink,容错,丢失,key,数据,checkpoint
From： https://www.cnblogs.com/T3h4t5-Blog90/p/18229314

flink状态和容错