什么是Exactly-Once一致性语义
Apache Spark的Exactly-once机制
Apache Flink的Exactly-once机制
Exactly-Once一致性语义
当任意条数据流转到某分布式系统中，如果系统在整个处理过程中对该任意条数据都仅精确处理一次，且处理结果正确，则被认为该系统满足Exactly-Once一致性。
以上仅是我个人对Exactly-once一致性语义的解释，相较于官方定义，显得更加通俗点，主要方便大家的理解。正如我的解释中描述的场景，在数据分析过程中需要满足精确一次处理的条件，这对于很多分布式多系统来说其实是个很大的考验。
因为分布式系统天生具有跨网络、多节点、高并发、高可用等特性，难免会出现节点异常、线程死亡、网络传输失败、并发阻塞等非可控情况，从而导致数据丢失、重复发送、多次处理等异常接踵而至。如何保持系统高效运行且数据仅被精确处理一次是很大的挑战。
分布式系统Exactly-Once的一致性保障，不是依靠某个环节的强一致性，而是要求系统的全流程均保持Exactly-Once一致性！！
Apache Spark的Exactly-Once机制
Apache Spark是一个高性能、内存级的分布式计算框架，在大数据领域中被广泛应用于离线分析、实时计算、数据挖掘等场景，因其采用独特的RDD数据模型及内存式计算，是海量数据分析和计算的利器之一。
实时场景下，Spark在整个流式处理中如何保证Exactly-Once一致性是重中之重。这需要整个系统的各环节均保持强一致性，包括可靠的数据源端(数据可重复读取、不丢失) 、可靠的消费端(Spark内部精确一次消费)、可靠的输出端(幂等性、事务)。
1. 数据源端
** 支持可靠的数据源接入(例如Kafka), 源数据可重读 **
- Spark Streaming内置的Kafka Direct API (KafkaUtils.createDirectStream)。实现精确Exactly-Once一致性语义。
Spark Streaming 自己管理offset（手动提交offset)，并保持到checkpoint中
Spark Streaming此时直接调用Kafka Consumer的API，自己管理维护offset(包括同步提交offset、保存checkpoint)，所以即使在重启情况下数据也不会重复。
val ssc = new StreamingContext(sc, Seconds(5))
// val topics = Map("spark" -> 2)
val kafkaParams = Map[String,String]{
"bootstrap.servers" -> "m1:9092,m2:9092,m3:9092",
"group.id" -> "spark",
"auto.offset.reset" -> "smallest"
}
// 直连方式拉取数据，这种方式不会修改数据的偏移量，需要手动的更新
val lines = kafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams)
Driver进程保持与Kafka通信，定期获取最新offset range范围，Executor进程根据offset range拉取kafka消息。因为Kafka本身offset就具有唯一特性，且Spark Streaming此时作为唯一的消费者，故全过程保持Exactly-once的一致性状态。注意: 如果程序崩溃，整个流可能会从earliest/latest处恢复重读，需考虑其他后续处理
(Spark-Kafka Direct API 流程示意图)
- Spark Streaming 基于Receiver的Kafka高级API，实现At least Once语义
基于Spark Streaming的Receiver模式，在Executor持续拉取kafka数据流
kafka数据存储到Executor内存和WAL(预写日志)中
WAL(预先日志)写入完成后，自动更新offset至zookeeper上
利用Spark本身的Receivers线程接收数据，内部调用Kafka高级消费API，不断触发batch消息拉取。获取的kafka数据在Executor本地存储，也可以启用WAL预写文件，将数据存储到第三方介质(HDFS)中。
val sparkConf = new SparkConf().setAppName("kafkaWordCount")
val ssc = new StreamingContext(sparkConf, Seconds(2))
ssc.checkpoint("checkpoint")
val topicMap = topics.split(",").map((_, numThreamds, toInt)).toMap
val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
此过程仅可实现At least once（至少一次）*，也就是说数据可能会被重复读取。即使理论上WAL机制可确保数据不丢失, 但是会存在消息写入WAL完成，但因其他原因无法及时更新offset至zookeeper的情况。此时kafka会重新发送offset，造成数据在Executor中多存储一份。
(Spark-Kafka 高级消费***者API 流程***示意图)
- 总结
(1) 高级消费者API需要启用Receiver线程消费Kafka数据，相较于第一种增加了开销，且无法直接实现并行读取，需要使用多个Kafka Dtstream 消费同一组然后union。
(2) 高级消费API在Executor本地和WAL存储两份数据<开启WAL不丢失机制>，而第一种Direct API仅在Executor中存储数据<offset存储到checkpoint中>
(3) 基于Kafka Direct API的方式，因Spark集成Kafka API直接管理offset，同时依托于Kafka自身特性，实现了Exactly-Once一致性语义。因此在生产中建议使用此种方式！！
2. Spark消费端
Spark的基本数据单元是一种被称作是RDD(分布式弹性数据集)的数据结构，Spark内部程序通过对RDD的进行一系列的transform和action操作，完成数据的分析处理。
基于RDD内存模型，启用多种一致性策略，实现Exactly-Once一致性。
-
RDD特性
(1) Spark的RDD是分布式、容错、不可变的数据集。其本身是只读的，不存储真实的数据，当结构更新或者丢失时可对RDD进行重建，RDD不会发生变化。
(2) 每个RDD都会有自己的Dependency RDD，即RDD的血脉机制。在开启 Checkpoint机制下，可以将RDD依赖保存到HDFS中。当RDD丢失或者程序出现问题，可以快速从血缘关系中恢复。因为记录了RDD的所有依赖过程，通过血脉追溯可重构计算过程且保证多次计算结果相同。
-
Checkpoint持久化机制 + WAL机制
(1) Spark的Checkpoint机制会在当前job执行完成后，再重新启动一个job，将程序中需要Checkpoint的RDD标记为MarkedForCheckpoint RDD, 且重新执行一遍RDD前面的依赖，完成后将结果保存到checkpoint中，并删除原先Dependency RDD依赖的血缘关系。同时可以将此次Checkpoint结果持久化到缓存中，便于后期快速恢复。利用Checkpoint的特性和高可用存储，保证RDD数据结果不丢失。
(2) 启用WAL预写文件机制。如果存在Driver或者Executor异常挂掉的场景，RDD结果或者jobs信息就会丢失，因此很有必要将此类信息持久化到WAL预写日志中，通过对元数据和中间数据存储备份，WAL机制可以防止数据丢失且提供数据恢复功能。
程序代码去重
如果实时流进入到Spark消费端已经存在重复数据，可以编写Spark程序代码进行****去重操作，实现Exactly-Once一致性。
(1) 内存去重。采用Hashset等数据结构，读取数据中类似主键等唯一性标识字段，在内存中存储并进行去重判断。
(2) 使用Redis Key去重。借助Redis的Hset等特殊数据类型，自动完成Key去重。
(3) DataFrame/SQL场景，使用group by/ over() window开窗等SQL函数去重
(4) 利用groupByKey等聚合算子去重
(5) 其他方法。。
3. 输出端
输出端保持Exactly-Once一致性，其输出源需要满足一定条件：
支持幂等写入、事务写入机制
- 幂等写入
首先解释一下幂等性，先看下百度百科上的定义:“ 幂等是一个数学与计算机学概念，常见于抽象代数中。在编程中一个幂等操作的特点******是其任意多次执行所产生的影响均与一次执行的影响相同。*****”*
结合语义可知，幂等写入就是多次写入会产生相同的结果，结果具有不可变性。在Spark中saveAsTextFile算子就是一种比较典型的幂等写入，也经常被用来作为数据的输出源。
此类型的写入方式，如果在消息中包含唯一主键，那么即使源头存在多条重复数据，在主键约束条件下也不会重复写入，从而实现Exactly-Once语义。
- 事务写入
相信大家对事务的概念都不陌生，在一个处理过程中的所有操作均需要满足一致性，即要不都发生，要不都不发生，常见于业务性、安全性要求比较高的场景，例如银行卡账户金额存取行为等，具有原子性、一致性、隔离性、持久性等四大特征。Spark读取Kafka数据需满足输出端的事务写入，则一般需生成一个唯一ID(可由批次号、时间、分区、offset等组合)，之后将该ID结合计算结果在同一个事务中写入目标源，提交和写入操作保持原子性，实现输出端的Exactly-Once语义。
