[spark streaming]Exactly-One的事务处理和不重复输出彻底掌握

缺乏、安全感 2022-05-13 03:50 94阅读 0赞

![Spark Streaming事务处理图][Spark Streaming]  
**Spark Streaming事务处理图分析：**   
1. Receiver不断地接收数据，收到数据后汇报给driver，driver收到数据后为了数据安全进行checkpoint，checkpoint中有：Configuration，Block MetaData，DStreamGraph,未处理完Job和等待中的Job   
2. job的执行完全基于spark core的调试模式   
3. Receiver接收数据通过BlockManager写入内存+磁盘或者通过WAL来保证数据的安全性，Receiver会等数据积累到一定量才写入，如果这个过程中数据丢失，如何恢复？   
a)Driver端的数据恢复：直接Driver的checkpoint的文件系统中将数据读取出来，而在内部其实是重新启动SparkContext，从新构建StreamingContext，恢复出元数据，再次产生RDD，再次提交到Spark集群。   
b)Receiver的重新恢复：Receiver在以前数据的基础上继续接收数据，曾经接收到的数据，通过WAL机制从磁盘中恢复回来。

**Exactly Once事务处理：**   
1. 数据的零丢失：必须有可要的数据来源和可靠的Receiver，且整个应用程序的metadata必须进行checkpoint，且通过WAL来保证数据安全，包括接收的数据和元数据本身，实际生产环境中数据来源一般都是Kafka，Receiver接收到来自于Kafka中的数据，默认存储的话是MEMONY\_AND\_DISK\_2.默认在执行计算的时候，他必须完成两台机器的容错之后，他才开始真正的进行计算。Receiver在接收数据如果崩溃的话，这个时候不会有数据丢失，此时没有完成默认副本的复制，Receiver恢复之后就可以重新接收。   
2. Spark Streaming1.3的时候，为了避免WAL的性能损失和实现Exactly Once而提供了Kafka Direct API，把Kafka作为文件存储系统！！！ Kafka是消息中间件，可以动态的接收数据，然后Spark Streaming又可以直接使用Direct API的方式直接操作Kafka，此时把Kafka作为文件存储系统，此时兼具流和文件系统的特性，直接对Kafka操作，Kafka又可以将数据存储一段时间，所以此时操作的时候直接操作Kafka数据中的Offset，这就可以确保数据肯定不会丢失，至此Spark Streaming + Kafka就构建了完美的流处理事件   
1). 数据不需要拷贝副本，   
2). 不需要WAL，因此没有性能损耗。   
3).Kafka比HDFS高效很多，因为Kafka中采用MEMORY COPY的方式)所有的Executor通过Kafka API直接消费数据   
如何解决不会重复读取数据的问题？直接管理Offset。所以不会重复消费数据，事务实现了。

**数据丢失及其具体的解决方式：**   
在Receiver收到数据且通过Driver的调度Executor开始计算数据的时候如果Driver突然崩溃，则此时Executor会被Kill掉，那么Executor中的数据就会丢失，此时就必须通过例如WAL的方式让所有的数据都通过例如HDFS的方式首先进行安全性容错处理，此时如果Executor中的数据丢失的话就可以通过WAL恢复回来。

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

**性能损失：**   
1.通过WAL的方式弊端是会极大的损伤Spark Streaming中Receivers接收数据的性能。Receiver接收Kafka的数据方式在实际的企业中使用不是那么多，一般都是直接使用Kafka读取数据。   
2.如果通过Kafka作为数据来源的话，Kafka中有数据，然后Receiver接受的时候又会有数据副本，这个时候其实是存储浪费。怎么解决？因为基于Zookeeper方式的话可以直接访问元数据信息，因此在处理的时候就可以将数据写入到内存数据库中（eg. SQLite），在处理的时候就去查该数据是否被处理过，如果处理过那就跳过即可。

**关于Spark Streaming数据输出多次重写及其解决方案：**   
1. 为什么会有这个问题？因为Spark Streaming在计算的时候基于Spark Core，Spark Core天生会做一下事情导致Spark Streaming的结果(部分)重复输出：   
1). Task重试；   
2). 慢任务推测;   
3). Stage重复;   
4). Job重试   
2. 具体解决方案：   
1).设置允许失败次数。Spark.task.maxFailures次数为1   
2). 设置spark.speculation为关闭状态(因为慢任务推测其实非常消耗性能，所以关闭后可以显著的提高Spark Streaming处理性能)。   
3).Spark Streaming on Kafka的话，Job失败后可以设置auto.offset.reset为”largest”的方式；这样就会自动进行恢复。

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

[Spark Streaming]: /images/20220513/df00af9f37304bb5834b8ca2c5b31c40.png