第4课：Spark Streaming的Exactly-Once的事务处理和不重复输出彻底掌握

前置知识：

1、事务的特征：1)、处理且仅被处理一次；2)、输出且只被输出一次

2、SparkStreaming进行事务处理有没有可能处理完全失败？

   这个可能性不大，因为Spark是批处理的方式来进行流处理，在SparkStreaming应用程序启动的时候，已经为应用程序分配了相关的资源，而且在调度的过程中可以动态的分配资源，所以除非整个集群所有的硬件都奔溃了，否则一般情况下都会被处理的。

3、SparkStreaming写程序的时候是基于Driver和Executor两部分

SparkStreaming架构流程：

1、SparkStreaming基本架构流程：

   1)、Receiver（不断的）接收到数据后汇报（把元数据）给Driver，2)、Driver在收到数据之后为了数据的安全性会进行CheckPoint，3)、Job的执行(在Executor中)：完全基于SparkCore的调度模式

   SparkStreaming基本架构流程图：

WAL（write ahead log）的机制：写数据的时候，先通过WAL机制写入文件系统中，然后存储到Executor，Executor在存储到磁盘或者内存中（这个是根据StorageLevel的设置） ，如果前面没有写成功的话，后面一定不会存储到Executor中，而不存储到Executor中的话，就不会汇报给Driver，数据就不会被处理了

Receiver接收的数据达到一定程度才会把数据存储到内存或者磁盘，当还没有积累到一定程度的时候，Executor或者Receiver奔溃了，这时数据就会丢失一点，

SparkStreaming：1、获取数据；2、产生作业，执行必须透过SparkContext

当出现奔溃的时候数据恢复的过程：

1)、Driver级别的恢复是直接从Driver进行checkpoint的文件系统中把数据读进来，而在内部是重新启动SparkContext（还有SparkContext），恢复出元数据再次产生RDD（恢复是基于上一次的job执行的），提交给集群

2)、Receiver的恢复是在以前数据的基础上接着去接收数据，曾经接收到的数据也会通过WAL机制从磁盘上恢复回来

Exactly Once的事务处理：

1)、数据零丢失：必须有可靠的数据来源和可靠的Receiver，且整个应用程序的metadata必须进行CheckPoint，且通过WAL来保证数据安全；(我们以数据来自Kafka为例，运行在Executor上的Receiver在接收到来自Kafka的数据时会向Kafka发送ACK确认收到信息并读取下一条信息，kafka会updateOffset来记录Receiver接收到的偏移，这种方式保证了在Executor数据零丢失。)

2)、Spark在1.3的时候为了避免WAL的性能损失和实现Exactly Once而提供了Kafka Direct API，把Kafka作为文件存储系统。此时兼具有流的优势和文件系统的优势，至此Spark Streaming+Kafka就构建了完美的流处理世界（1，数据不需要拷贝副本；2，不需要WAL对性能的损耗；3，Kafka使用ZeroCopy比HDFS更高效）。所有的Executors通过Kafka API直接消息数据，直接管理Offset，所以也不会重复消费数据

数据丢失及其具体解决方式：

在Receiver收到数据且通过Driver的调度Executor开始计算数据的时候，如果Driver突然奔溃，则此时Executor会被kill掉，那么Executor中的数据就会丢失(如果没有进行WAL的操作)。

解决方式：此时就必须通过例如WAL的方式，让所有的数据都通过例如HDFS的方式首先进行安全性容错处理。此时如果Executor中的数据丢失的话，就可以通过WAL恢复回来(这种方式的弊端是通过WAL的方式会极大额损伤SparkStreaming中Receivers接收数据的性能)

数据重复读取的情况：

基于Kafka的情况下，Receiver收到数据且保存到了HDFS等持久化引擎但是没有来得及进行updateOffsets，此时Receiver崩溃后重新启动就会通过管理Kafka的ZooKeeper中元数据再次重复读取数据，但是此时SparkStreaming认为是成功的，但是Kafka认为是失败的（因为没有更新offset到ZooKeeper中），此时就会导致数据重新消费的情况。

解决方式：以Receiver基于ZooKeeper的方式，当读取数据时去访问Kafka的元数据信息，在处理代码中例如foreachRDD或transform时，将信息写入到内存数据库中(memorySet)，在计算时读取内存数据库信息，判断是否已处理过，如果以处理过则跳过计算。这些元数据信息可以保存到内存数据结构或者memsql，sqllite中(如果通过Kafka作为数据来源的话，Kafka中有数据，然后Receiver接收的时候又会有数据副本，这个时候其实是存储资源的浪费)

数据输出多次重写

为什么会有这个问题，因为Spark Streaming在计算的时候基于Spark Core，Spark Core天生会做以下事情导致Spark Streaming的部分结果重复输出(例如数据输出后，该Task的后续程序发生错误，而任务发生错误，Spark Core会进入如下程序)：

  Task重试；慢任务推测(两个相同任务可能会同时执行)，Stage重复；Job重试；

解决方式：

设置spark.task.maxFailures次数为1；

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

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

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

备注：

1、DT大数据梦工厂微信公众号DT_Spark 
2、IMF晚8点大数据实战YY直播频道号：68917580
3、新浪微博: http://www.weibo.com/ilovepains