本文共 4358 字，大约阅读时间需要 14 分钟。
本文主要整理实时组件（SparkStreaming VS Flink）容错及语义
内容如下：消息系统系统一般有以下的语义：
比较典型的消息系统 kafka，其语义的说明可参考：https://blog.csdn.net/super_wj0820/article/details/98886110
此文详细说明了 Kafka 不同版本做出的 语义 保证
应该说，Exactly-Once 语义是业务想要的最理想效果，下文主要考量 Spark Streaming 和 Flink 如何在整个任务链上保证 Exactly-Once 语义
一个流式计算处理程序，从广义上说包括三个步骤：
如果流处理程序需要实现Exactly-Once语义，那么每一个步骤都要保证 Exactly-Once
SparkStreaming 和 Flink 作为流式计算执行引擎，内部天然支持 Exactly-Once 语义
因 SparkStreaming 内部天然支持 Exactly-Once 语义，所以只需考虑 SparkStreaming 连接上下游组件时实现 Exactly-Once
不同的数据源提供不同的保证
如HDFS中的数据源，直接支持Exactly-Once语义；如使用基于Kafka Direct API从Kafka获取数据，也能保证 Exactly-Once
但是 通过 Kafka Direct API 消费数据时，不会提交offset，需要对 offset 进行管理，以便在任务 启停 和 crash 恢复后实现 Exactly-Once 语义
Spark Streaming 的 checkpoints 是最基本的存储状态信息的方式，一般是保存在HDFS中
但是最大的问题是如果streaming程序升级的话，checkpoints的数据无法使用，也就丢失了offset信息，所以几乎没人使用
此方式也并不会向 Zookeeper 或 Broken 提交offset，不便于对 消费情况做监控
因此，建议不采用此方案
enable.auto.commit=true
一但 consumer 挂掉，就会导致数据丢失或重复消费，并且 offset 不可控
因此，建议彻底放弃使用这种方式
Kafka 0.10+ 版本，在确保处理完输出Sink后，手动提交offset（commitAsync）
备注：sparkStreaming消费kafka-0.8方式：direct方式（存储offset到zookeeper）（https://www.cnblogs.com/niutao/p/10547594.html）
stream.foreachRDD { rdd => val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges // some time later, after outputs have completed stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges) } 此时参数修改为 enable.auto.commit=false
理论上来说，将 commitAsync 操作 和 输出Sink操作放在一个 事务中，可以实现 Exactly-Once
但是 offset 提交并不是 可靠的操作，消费完后写offset到zk失败，这个状态consumer客户端是感知不到的，二者并没有类似TCP的ack机制（参考：https://www.zhihu.com/question/334249637/answer/745816753）（失败概率极小）
因此，事务中手动提交 offset 可基本实现 Exactly-Once（推荐）
将 offset 存放在第方三储中，包括 RDBMS、Redis、ZK、ES 甚至 Kafka 中
若消费数据存储在带事务的组件上，则强烈推荐将 offset 存储在一起，借助事务实现 Exactly-once 语义
但是这种方式外界同样不便观察消费进度
ps：对于自带幂等性的下游组件，天生可以自动将 at least once 转化为 exactly once
由 3.1 节可以看出，要实现 exactly once 语义，需将 输出Sink操作 和 offset 提交放在一个事务中
除此之外，参考 1节 中kafka的语义说明，需要配置 kafka 的 ack=all，以保证发送的数据到达 kafka服务端
参考：
http://shzhangji.com/cnblogs/2017/08/01/how-to-achieve-exactly-once-semantics-in-spark-streaming/ https://www.cnblogs.com/littlemagic/p/11440239.html https://www.jianshu.com/p/885505daab29 https://blog.csdn.net/wangpei1949/article/details/89277490 https://blog.csdn.net/qq_38976805/article/details/96210028 https://www.jianshu.com/p/d2a61be73513 https://cloud.tencent.com/developer/news/327516
因 Flink 内部天然支持 Exactly-Once 语义，所以只需考虑 SparkStreaming 连接上下游组件时实现 Exactly-Once
但与 SparkStreaming 不同的是，SparkStreaming 是微批处理，每个 rdd 处理完成往 Sink 组件发送数据的同时 提交 offset，但是 Flink 是数据逐条处理，每处理完一条数据就提交offset不现实
所以 Flink 在实现 End-to-End Exactly-Once语义时，Flink采用 Two phase commit 来解决这个问题
Flink 提交offset有两种方式：一是依赖 kafka 自身属性，定时自动提交（参考3.1.2）；二是依赖 Checkpoint 机制，每次 Checkpoint 完成向外提交
自动提交 offset 无关管理 offset，所以几乎不做考虑
Flink 在 Checkpoint 中保存消费进度，即offset，那么 Checkpoint 只要保证 输出 Sink 时的 Exactly-Once语义，并在 Checkpoint 中同步提交 offset，即可实现 End-to-End Exactly-Once语义
Flink 为实现 Exactly-Once语义，提出了 two phase commit
Phase 1: Pre-commit
Flink 的 JobManager 向 source 注入 checkpoint barrier 以开启这次 snapshot
barrier 从 source 流向 sink
每个进行 snapshot 的算子成功 snapshot 后,都会向 JobManager 发送 ACK
当 sink 完成 snapshot 后, 向 JobManager 发送 ACK 的同时向 kafka 进行 pre-commit
Phase 2:Commit
当 JobManager 接收到所有算子的 ACK 后，就会通知所有的算子这次 checkpoint 已经完成（同步提交上游Topic的offset）
Sink接收到这个通知后, 就向 kafka 进行 commit, 正式把数据写入到kafka
4.2.1 不同阶段 fail over 的 recovery 举措:
在pre-commit前fail over，系统恢复到最近的checkponit
在pre-commit后，commit前fail over，系统恢复到刚完成pre-commit时的状态
但是如果 Flink 执行引擎下游是 Kafka，因为 kafka 0.11 之后，自带事务实现幂等操作，类似 Flink 的 two phase commit
补充：
Flink的two phase commit实现 ---- 抽象类TwoPhaseCommitSinkFunction TwoPhaseCommitSinkFunction有4个方法: 1. beginTransaction() 开启事务.创建一个临时文件.后续把原要写入到外部系统的数据写入到这个临时文件 2. preCommit() flush并close这个文件,之后便不再往其中写数据.同时开启一个新的事务供下个checkponit使用 3. commit() 把pre-committed的临时文件移动到指定目录 4. abort() 删除掉pre-committed的临时文件
参考：
https://www.cnblogs.com/tuowang/p/9025266.html https://www.cnblogs.com/zzjhn/p/11525086.html