98、SparkStreaming之容错机制以及事务语义详解

Spark Streaming容错机制

要理解Spark Streaming提供的容错机制&＃xff0c;先回忆一下Spark RDD的基础容错语义

RDD的基本容错原理.png

1. RDD&＃xff0c;Ressilient Distributed Dataset&＃xff0c;是不可变的、确定的、可重新计算的、分布式的数据集。每个RDD都会记住确定好的计算操作的血缘关系&＃xff0c;&＃xff08;val lines &＃61; sc.textFile(hdfs file); val words &＃61; lines.flatMap(); val pairs &＃61; words.map(); val wordCounts &＃61; pairs.reduceByKey()&＃xff09;这些操作应用在一个容错的数据集上来创建RDD。
2. 如果因为某个Worker节点的失败&＃xff08;挂掉、进程终止、进程内部报错&＃xff09;&＃xff0c;导致RDD的某个partition数据丢失了&＃xff0c;那么那个partition可以通过对原始的容错数据集应用操作血缘&＃xff0c;来重新计算出来。
3. 所有的RDD transformation操作都是确定的&＃xff0c;最后一个被转换出来的RDD的数据&＃xff0c;一定是不会因为Spark集群的失败而丢失的。

Spark操作的通常是容错文件系统中的数据&＃xff0c;比如HDFS。因此&＃xff0c;所有通过容错数据生成的RDD也是容错的。然而&＃xff0c;对于Spark Streaming来说&＃xff0c;这却行不通&＃xff0c;因为在大多数情况下&＃xff0c;数据都是通过网络接收的&＃xff08;除了使用fileStream数据源&＃xff09;。要让Spark Streaming程序中&＃xff0c;所有生成的RDD&＃xff0c;都达到与普通Spark程序的RDD&＃xff0c;相同的容错性&＃xff0c;接收到的数据必须被复制到多个Worker节点上的Executor内存中&＃xff0c;默认的复制因子是2。

基于上面说的&＃xff0c;在出现失败的事件时&＃xff0c;有两种数据需要被恢复&＃xff1a;

1. 数据接收到了&＃xff0c;并且已经复制过——这种数据在一个Worker节点挂掉时&＃xff0c;是可以继续存活的&＃xff0c;因为在其他Worker节点上&＃xff0c;还有它的一份副本。
2. 数据接收到了&＃xff0c;但是正在缓存中&＃xff0c;等待复制的——因为还没有复制该数据&＃xff0c;因此恢复它的唯一办法就是重新从数据源获取一份。

此外&＃xff0c;还有两种失败是我们需要考虑的&＃xff1a;

1. Worker节点的失败——任何一个运行了Executor的Worker节点的挂掉&＃xff0c;都会导致该节点上所有在内存中的数据都丢失。如果有Receiver运行在该Worker节点上的Executor中&＃xff0c;那么缓存的&＃xff0c;待复制的数据&＃xff0c;都会丢失。
2. Driver节点的失败——如果运行Spark Streaming应用程序的Driver节点失败了&＃xff0c;那么显然SparkContext会丢失&＃xff0c;那么该Application的所有Executor的数据都会丢失。

Spark Streaming容错语义的定义

流式计算系统的容错语义&＃xff0c;通常是以一条记录能够被处理多少次来衡量的。有三种类型的语义可以提供&＃xff1a;

1. 最多一次&＃xff1a;每条记录可能会被处理一次&＃xff0c;或者根本就不会被处理。可能有数据丢失。
2. 至少一次&＃xff1a;每条记录会被处理一次或多次&＃xff0c;这种语义比最多一次要更强&＃xff0c;因为它确保零数据丢失。但是可能会导致记录被重复处理几次。
3. 一次且仅一次&＃xff1a;每条记录只会被处理一次——没有数据会丢失&＃xff0c;并且没有数据会处理多次。这是最强的一种容错语义。

Spark Streaming的基础容错语义

在Spark Streaming中&＃xff0c;处理数据都有三个步骤&＃xff1a;

1. 接收数据&＃xff1a;使用Receiver或其他方式接收数据。
2. 计算数据&＃xff1a;使用DStream的transformation操作对数据进行计算和处理。
3. 推送数据&＃xff1a;最后计算出来的数据会被推送到外部系统&＃xff0c;比如文件系统、数据库等。

如果应用程序要求必须有一次且仅一次的语义&＃xff0c;那么上述三个步骤都必须提供一次且仅一次的语义。每条数据都得保证&＃xff0c;只能接收一次、只能计算一次、只能推送一次。Spark Streaming中实心这些语义的步骤如下&＃xff1a;

1. 接收数据&＃xff1a;不同的数据源提供不同的语义保障。
2. 计算数据&＃xff1a;所有接收到的数据一定只会被计算一次&＃xff0c;这是基于RDD的基础语义所保障的。即使有失败&＃xff0c;只要接收到的数据还是可访问的&＃xff0c;最后一个计算出来的数据一定是相同的。
3. 推送数据&＃xff1a;output操作默认能确保至少一次的语义&＃xff0c;因为它依赖于output操作的类型&＃xff0c;以及底层系统的语义支持&＃xff08;比如是否有事务支持等&＃xff09;&＃xff0c;但是用户可以实现它们自己的事务机制来确保一次且仅一次的语义。

接收数据的容错语义

1. 基于文件的数据源
   如果所有的输入数据都在一个容错的文件系统中&＃xff0c;比如HDFS&＃xff0c;Spark Streaming一定可以从失败进行恢复&＃xff0c;并且处理所有数据。这就提供了一次且仅一次的语义&＃xff0c;意味着所有的数据只会处理一次。

2. 基于Receiver的数据源
   对于基于Receiver的数据源&＃xff0c;容错语义依赖于失败的场景和Receiver类型。
   可靠的Receiver&＃xff1a;这种Receiver会在接收到了数据&＃xff0c;并且将数据复制之后&＃xff0c;对数据源执行确认操作。如果Receiver在数据接收和复制完成之前&＃xff0c;就失败了&＃xff0c;那么数据源对于缓存的数据会接收不到确认&＃xff0c;此时&＃xff0c;当Receiver重启之后&＃xff0c;数据源会重新发送数据&＃xff0c;没有数据会丢失。

    

   

   可靠Receiver的原理.png

    

   不可靠的Receiver&＃xff1a;这种Receiver不会发送确认操作&＃xff0c;因此当Worker或者Driver节点失败的时候&＃xff0c;可能会导致数据丢失。
   不同的Receiver&＃xff0c;提供了不同的语义。如果Worker节点失败了&＃xff0c;那么使用的是可靠的Receiver的话&＃xff0c;没有数据会丢失。使用的是不可靠的Receiver的话&＃xff0c;接收到&＃xff0c;但是还没复制的数据&＃xff0c;可能会丢失。如果Driver节点失败的话&＃xff0c;所有过去接收到的&＃xff0c;和复制过缓存在内存中的数据&＃xff0c;全部会丢失。
   要避免这种过去接收的所有数据都丢失的问题&＃xff0c;Spark从1.2版本开始&＃xff0c;引入了预写日志机制&＃xff0c;可以将Receiver接收到的数据保存到容错存储中。如果使用可靠的Receiver&＃xff0c;并且还开启了预写日志机制&＃xff0c;那么可以保证数据零丢失。这种情况下&＃xff0c;会提供至少一次的保障。&＃xff08;Kafka是可以实现可靠Receiver的&＃xff09;

从Spark 1.3版本开始&＃xff0c;引入了新的Kafka Direct API&＃xff0c;可以保证&＃xff0c;所有从Kafka接收到的数据&＃xff0c;都是一次且仅一次。基于该语义保障&＃xff0c;如果自己再实现一次且仅一次语义的output操作&＃xff0c;那么就可以获得整个Spark Streaming应用程序的一次且仅一次的语义。

输出数据的容错语义

output操作&＃xff0c;比如foreachRDD&＃xff0c;可以提供至少一次的语义。那意味着&＃xff0c;当Worker节点失败时&＃xff0c;转换后的数据可能会被写入外部系统一次或多次。对于写入文件系统来说&＃xff0c;这还是可以接收的&＃xff0c;因为会覆盖数据。但是要真正获得一次且仅一次的语义&＃xff0c;有两个方法&＃xff1a;

1. 幂等更新&＃xff1a;多次写操作&＃xff0c;都是写相同的数据&＃xff0c;例如saveAs系列方法&＃xff0c;总是写入相同的数据。
2. 事务更新&＃xff1a;所有的操作都应该做成事务的&＃xff0c;从而让写入操作执行一次且仅一次。给每个batch的数据都赋予一个唯一的标识&＃xff0c;然后更新的时候判定&＃xff0c;如果数据库中还没有该唯一标识&＃xff0c;那么就更新&＃xff0c;如果有唯一标识&＃xff0c;那么就不更新。

dstream.foreachRDD { (rdd, time) &＃61;>rdd.foreachPartition { partitionIterator &＃61;>val partitionId &＃61; TaskContext.get.partitionId()val uniqueId &＃61; generateUniqueId(time.milliseconds, partitionId)// partitionId和foreachRDD传入的时间&＃xff0c;可以构成一个唯一的标识}
}