本讲内容：

a. Exactly Once
b. 输出不重复

注：本讲内容基于Spark 1.6.1版本（在2016年5月来说是Spark最新版本）讲解。

上节回顾:

上节课通过案例透视了Spark Streaming Job架构和运行机，并结合源码进行了详细解说；同时也了解了Spark Streaming Job的容错机制，包括 Executor 与 Driver两方面的容错机制。

也就是说Job的事务处理，主要是在Executor 与 Driver两个应用中展开

开讲

首先，我们必须知道什么是事务及其一致性？

事务应该具有4个属性：原子性、一致性、隔离性、持久性。这四个属性通常称为ACID特性。

原子性（atomicity）。一个事务是一个不可分割的工作单位，事务中包括的诸操作要么都做，要么都不做。

一致性（consistency）。事务必须是使数据库从一个一致性状态变到另一个一致性状态。一致性与原子性是密切相关的。

隔离性（isolation）。一个事务的执行不能被其他事务干扰。即一个事务内部的操作及使用的数据对并发的其他事务是隔离的，并发执行的各个事务之间不能互相干扰。

持久性（durability）。持久性也称永久性（permanence），指一个事务一旦提交，它对数据库中数据的改变就应该是永久性的。接下来的其他操作或故障不应该对其有任何影响。

以银行转帐为例，A客户向B客户转账一次（假如此次转账1万元），正常情况下A客户的账户里只会被扣除一次且金额为一万元，B客户的账户也只会收到A客户转给的一次钱且金额同样是一万元，这就是事务及其一致性的具体体现，也就是说数据会被处理且会被正确的处理一次。

然而， Spark Streaming的事务处理和上述事例中讲的事务及其一致性有所不同；Spark Streaming的事务关注的是某个Job执行的一致性。

本讲将从事务视角为大家探索Spark Streaming架构机制

Spark Streaming应用程序启动，会分配资源，除非整个集群硬件资源奔溃，一般情况下都不会有问题。Spark Streaming程序分成而部分，一部分是Driver，另外一部分是Executor。Receiver接收到数据后不断发送元数据给Driver，Driver接收到元数据信息后进行CheckPoint处理。其中CheckPoint包括：Configuration（含有Spark Conf、Spark Streaming等配置信息）、Block MetaData、DStreamGraph、未处理完和等待中的Job。当然Receiver可以在多个Executor节点的上执行Job，Job的执行完全基于SparkCore的调度模式进行的。

　Executor只有函数处理逻辑和数据，外部InputStream流入到Receiver中通过BlockManager写入磁盘、内存、WAL进行容错。WAL先写入磁盘然后写入Executor中，失败可能性不大。如果1G数据要处理，Executor一条一条接收，Receiver接收数据是积累到一定记录后才会写入WAL，如果Receiver线程失败时，数据有可能会丢失。
　
　　Driver处理元数据前会进行CheckPoint，Spark Streaming获取数据、产生作业，但没有解决执行的问题，执行一定要经过SparkContext。Driver级别的数据修复从Driver CheckPoint中需要把数据读入，在其内部会重新构建SparkContext、StreamingContext、SparkJob，再提交Spark集群运行。Receiver的重新恢复时会通过磁盘的WAL从磁盘恢复过来。

　Spark Streaming和Kafka结合不会出现WAL数据丢失的问题，Spark Streaming必须考虑外部流水线的方式处理。

　上面的图例很好的解释了怎么能完成完整的语义、事务的一致性，保证数据的零丢失，Exactly Once的事务处理？
　
　　a、怎么保证数据零丢失？
　　
　　必须要有可靠的数据来源和可靠的Receiver、整个应用程序的MetaData必须进行CheckPoint、通过WAL来保证数据安全（生产环境下Receiver接收Kafka的数据，默认情况下会在Executor中存在二份数据，且默认情况下必须二份数据备份后才进行计算；如果Receiver接收数据时崩溃，没有拷贝副本，此时会重新从Kafka中进行拷贝，拷贝的依据是zookeeper元数据）。
　　
　　大家可以将Kafka看作是一个简单的文件存储系统，在Executor中Receiver确定受到Kafka的每一条记录后进行Replication到其他Executor成功后会通过ack向Kafka发送确认收到的信息并继续从Kafka中读取下一条信息。
　　
　　b、Driver容错如下图所示：

再次思考数据在哪些地方可能丢失？

　　数据丢失的主要场景如下：

　　在Receiver收到数据且通过Driver的调度，Executor开始计算数据的时候如果Driver突然崩溃（导致Executor会被Kill掉），此时Executor会被Kill掉，那么Executor中的数据就会丢失，此时就必须通过例如WAL机制让所有的数据通过类似HDFS的方式进行安全性容错处理，从而解决Executor被Kill掉后导致数据丢失可以通过WAL机制恢复回来。

下面需要考虑二个很重要的场景：

数据的处理怎么保证有且仅有被处理一次？

数据零丢失并不能保证Exactly Once，如果Receiver接收且保存起来后没来得及更新updateOffsets时，就会导致数据被重复处理。

更详细的说明数据重复读取的场景：

　　在Receiver收到数据且保存到了hdfs时Receiver奔溃，此时持久化引擎没有来得及进行updateOffset，Receiver重新启动后就会从管理Kafka的ZooKeeper中再次读取元数据从而导致重复读取元数据；从SparkStreaming来看是成功的，但是Kafka认为是失败的（因为Receiver奔溃时没有及时更新offsets到ZooKeeper中）重新恢复时会重新消费一次，此时会导致数据重新消费的情况。

性能补充：

a、通过WAL方式保证数据不丢失，但弊端是通过WAL方式会极大的损伤SparkStreaming中的Receiver接收数据的性能（现网生产环境通常会Kafka direct api直接处理）。

b、需要注意到是：如果通过Kafka作为数据来源的话，Kafka中有数据，然后Receiver接受数据的时候又会有数据副本，这个时候其实是存储资源的浪费。（重复读取数据解决办法，读取数据时可以将元数据信息放入内存数据库中，再次计算时检查元数据是否被计算过）。

　　Spark1.3的时候为了避免WAL的性能损失和实现Exactly Once而提供了Kafka direct api，把Kafka作为文件存储系统！！！此时Kafka兼具有流的优势和文件系统的优势，至此，Spark Streaming+Kafka就构建了完美的流处理世界！！！

　　数据不需要copy副本，不需要WAL性能损耗，不需要Receiver，而直接通过kafka direct api直接消费数据，所有的Executors通过kafka api直接消费数据，直接管理offset，所以也不会重复消费数据；事务实现啦！！！

关于Spark Streaming数据输出多次重写及其解决方案

a、为什么会有这个问题，因为Spark Streaming在计算的时候基于Spark Core，Spark Core天生会做以下事情导致Spark Streaming的结果（部分）重复输出：

Task重试；

慢任务推测

Stage重复；

Job重试；

b、具体解决方案：

设置spark.task.maxFailures次数为1；最大允许失败的次数，设为1就没有task、stage、job等的重试；

设置spark.speculation为关闭状态（因为慢任务推测其实非常消耗性能，所以关闭后可以显著提高Spark Streaming处理性能）

Spark Streaming on Kafka的话，Job失败会导致任务失败，Job失败后可以设置auto.offset.reset为“largest”的方式；

最后再次强调

　　可以通过transform和foreachRDD基于业务逻辑代码进行逻辑控制来实现数据不重复消费和输出不重复！这二个方法类似于spark的后门，可以做任意想象的控制操作！

备注：
1、DT大数据梦工厂微信公众号DT_Spark
2、Spark大神级专家：王家林
3、新浪微博: http://www.weibo.com/ilovepains