Flink的状态一致性

1 状态的一致性

1.1 一致性级别

流处理操作一般分为at-most-once，at-least-once和exactly-once这3个级别。

at-most-once：至多一次，发生故障恢复后数据可能丢失

at-least-once：至少一次，发生故障恢复后数据可能多算，绝对不会少算

exactly-once：精确一次，发生故障恢复后数据不会丢失也不会多算

1.2 端到端的状态一致性

1.2.1 Spark Streaming的exactly-once

对于 Spark Streaming 任务，设置了 checkpoint，如果发生故障并重启，可以从上次 checkpoint 之处恢复，但是这个行为只能使得数据不丢失，可能会重复处理，不能做到恰一次处理语义。如果 Spark Streaming 消费的数据源是 kafka ，那么使用 direct Stream 的方式自己维护 offset 到 zookeeper或者其它外部系统，每次提交完结果之后再提交 offset，这样故障恢复重启可以利用上次提交的 offset 恢复，保证数据不丢失。但是假如故障发生在提交结果之后、提交 offset 之前会导致数据多次处理，就需要保证处理结果多次输出不影响正常的业务。

所以如果要保证数据恰一次处理语义，那么结果输出和 offset 提交必须在一个事务内完成。在这里有以下两种做法：

（1）使用repartition(1) 将输出的 partition设置为1，那就可以利用事务操作

Dstream.foreachRDD(rdd=>{rdd.repartition(1).foreachPartition(partition=>{    //    开启事务partition.foreach(each=>{//        提交数据})    //  提交事务})})

（2）将结果和offset一起提交。这样提交结果和提交 offset 就是一个操作完成，不会数据丢失，也不会重复处理。故障恢复的时候可以利用上次提交结果带的 offset。

1.2.2 Flink的exactly-once

Flink使用checkpoint保证其内部的exactly-once，但是我们的应用还包含了数据源和输出，每个组件都只是保证了自己的一致性，所以端到端级别的一致性取决于所有组件中一致性最弱的组件。

要满足端到端的状态一致性需要满足以下几点：

（1）source：需要外部数据源可以重新设置数据的读取位置

（2）内部：通过checkpoint保证内部的一致性

（3）sink：从故障恢复时，数据不会重复写入到外部系统

如果sourced端是kafka的话，可以轻松实现重设读取位置，Flink内部通过checkpoint就能保证内部一致性，较为复杂的是在sink端不能重复写入，有两种具体的实现方式：幂等写入和事务性写入。

幂等写入：重复执行多次操作，但是重复执行就只会导致一次结果更新，重复执行不起作用。幂等写入要求外部的数据库必须支持幂等写入，像往es和文件里面追加写入是不行的，像redis和mysql定义了key之后也不一定是真正的exectly-once，保证的是最终的状态一次性，在中间短暂的恢复中是有短暂的状态不一致的，因为有些中间状态要重复写入

事务写入：构建事务写入外部系统，构建的事务对应checkpoint，到checkpoint真正完成的时候才把所有对应的结果写入到sink系统。对于事务性写入，又有两种实现方式：预写日志和两阶段提交。GenericWriteAheadSink模板类和TwoPhaseCommitSinkFunction 接口，可以方便地实现这两种方式的事务性写入

1.2.3 预写日志

把所有要写入的数据保存成sink的一个状态，相当于在sink做缓存了，收到checkpoint完成的通知的时候再一次性写入sink系统。再往外部系统写入的时候一批写入如果写到一半的时候挂了怎么办，不能保证。不能严格意义达到精确一次性。DataStream API提供了GenericWriteAheadSink模板类

1.2.4 两阶段提交

两阶段提交能真正做到exectly-once。前面说的预写入日志是对两次checkpoint之间的所有数据直接缓存在sink任务里面，最后收到checkpoint通知的时候一批直接写入。两阶段事务提交还是正常的来一个写一个，只不过和外部系统开启了一个事务，是在事务里面提交的，假如中间某个挂了就全部回滚，等到checkpoint完成的时候就真正把这个事务提交。

2 端到端状态一致性

端到端的状态一致性的实现，需要每一个组件都实现。以Kafka-Flink-Kafka的端到端的数据管道系统为例：

①source：kafka consumer作为source，可以将偏移量保存下来，如果任务出现了故障，恢复的时候可以由连接器重置偏移量，重新消费数据，即可保证一致性。②内部：利用checkpoint机制，把状态存盘，发生故障的时候可以恢复，保证内部的状态一致性。③sink ： kafka producer作为sink，采用两阶段提交 sink，需要实现一个 TwoPhaseCommitSinkFunction

Flink是由JobManager协调各个TaskManager进行checkpoint存储， Kafka-Flink-Kafka的流程如下：

checkpoint 启动，JobManager 会将检查点分界线barrier注入数据流；barrier往后传递下去，每个算子遇到barrier会对当前的状态做个快照，保存到状态后端。source就是当前的offset作为状态保存起来，checkpoint恢复时重新提交偏移量，从上次保存的位置开始重新消费数据。barrier一直传递带sink，当遇到 barrier 时，把状态保存到状态后端，并开启新的预提交事务，sink 任务首先把数据写入外部 kafka，这些数据都属于预提交的事务。当所有算子任务的快照完成，也就是这次的 checkpoint 完成时，JobManager 会向所有任务发通知，确认这次 checkpoint 完成。当sink 任务收到确认通知，就会正式提交之前的事务，kafka 中未确认的数据就改为“已确认”，数据就真正可以被消费了。

两阶段提交步骤：①预提交：第一条数据来了之后，开启一个 kafka 的事务（transaction），正常写入 kafka 分区日志但标记为未提交②jobmanager 触发 checkpoint 操作，barrier 从 source 开始向下传递，遇到 barrier 的算子将状态存入状态后端，并通知 jobmanager③ sink 连接器收到 barrier，保存当前状态，存入 checkpoint，通知 jobmanager，并开启下一阶段的事务，用于提交下个检查点的数据④jobmanager 收到所有任务的通知，发出确认信息，表示 checkpoint 完成⑤sink 任务收到 jobmanager 的确认信息，正式提交这段时间的数据⑥外部kafka关闭事务，提交的数据可以正常消费了。