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

心已赠人 2022-08-22 00:14 120阅读 0赞

**本期内容**

1 Exactly-Once事务处理

2 输出不重复的解决办法

**1 Exactly-Once事务处理**

**1）什么是Exactly-Once事务？**

数据仅处理一次并且仅输出一次，这样才是完整的事务处理。

以银行转帐为例，A用户转账给B用户，B用户可能收到多笔钱，保证事务的一致性，也就是说事务输出，能够输出且只会输出一次，即A只转一次，B只收一次。

**2）从事务视角解密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的调度模式进行的。

![SouthEast][]

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从磁盘恢复过来。

**3）数据可能丢失的情况及通常的解决方式**

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

此时就必须通过例如WAL的方式，让所有的数据都通过例如HDFS的方式首先进行安全性容错处理。此时如果Executor中的数据丢失的话，就可以通过WAL恢复回来。

注意：这种方式的弊端是通过WAL的方式会极大额损伤SparkStreaming中Receivers接收数据的性能。

**       4）Exactly-Once事务处理如何解决数据零丢失？**

解决数据零丢失，必须有可靠的数据来源和可靠的Receiver，且整个应用程序的metadata必须进行checkpoint，且通过WAL来保证数据安全。

我们以数据来自Kafka为例，运行在Executor上的Receiver在接收到来自Kafka的数据时会向Kafka发送ACK确认收到信息并读取下一条信息，Kafka会通过updateOffset来记录Receiver接收到的偏移，这种方式保证了在Executor数据零丢失。在Driver端，定期进行checkpoint操作，出错时从Checkpoint的文件系统中把数据读取进来进行恢复，内部会重新构建StreamingContext(也就是构建SparkContext)并启动，恢复出元数据metedata，再次产生RDD，恢复的是上次的Job，然后再次提交到集群执行。

![SouthEast 1][]

**5）Exactly-Once事务处理如何解决数据重复读取？**

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

以Receiver基于ZooKeeper的方式，当读取数据时去访问Kafka的元数据信息，在处理代码中例如foreachRDD或transform时，将信息写入到内存数据库中(memorySet)，在计算时读取内存数据库信息，判断是否已处理过，如果以处理过则跳过计算。这些元数据信息可以保存到内存数据结构或者memsql，sqllite中。

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

**2 输出不重复的解决办法**

**1）为什么会有数据输出多次重写这个问题？**

因为Spark Streaming在计算的时候基于Spark Core天生会做以下事情导致Spark Streaming的结果(部分)重复输出：

*  Task重试
 *  慢任务推测
 *  Stage重试
 *  Job重试

**2）具体解决方案？**

*  设置spark.task.maxFailures次数为1，这样就不会有Task重试了。
 *  设置spark.speculation为关闭状态，就不会有慢任务推测了，因为慢任务推测非常消耗性能，所以关闭后可以显著提高Spark Streaming处理性能。
 *  Spark Streaming On Kafka的话，Job失败后可以设置Kafka的参数auto.offset.reset为largest方式。

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

[SouthEast]: /images/20220724/7c74974a9faf4e6d920992f0111a7a52.png
[SouthEast 1]: /images/20220724/3fbbe5dcb7234ebeb58007db1e94d8ea.png