Spark Streaming之容错机制

迷南。 2022-12-29 04:48 140阅读 0赞

### 目录 ###

*   *  一、容错机制的背景
     *  二、Spark Streaming 容错语义的定义
     *  三、接收数据的容错语义
     *  四、输出数据的容错语义
     *  五、storm的容错语义

## 一、容错机制的背景 ##

**要理解Spark Streaming提供的容错机制，先回忆一下Spark RDD的基础容错语义：**

*  RDD，Ressilient Distributed Dataset，是不可变的、确定的、可重新计算的、分布式的数据集。每个 RDD都会记住确定好的计算操作的血缘关系，这些操作应用在 一个容错的数据集上来创建RDD。

val lines = sc.textFile(hdfs file); 
    val words = lines.flatMap(); 
    val pairs = words.map(); 
    val wordCounts = pairs.reduceByKey()

*  如果因为某个Worker节点的失败（挂掉、进程终止、进程内部报错），导致RDD的某个partition数据丢失 了，那么那个partition可以通过对原始的容错数据集应用操作血缘，来重新计算出来。
 *  所有的RDD transformation操作都是确定的，最后一个被转换出来的RDD的数据，一定是不会因为Spark集 群的失败而丢失的。

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

**基于上述理论，在出现失败的事件时，有两种数据需要被恢复：**

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

**此外，还有两种失败是我们需要考虑的：**

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

## 二、Spark Streaming 容错语义的定义 ##

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

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

**在Spark Streaming中，处理数据都有三个步骤：**

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

如果应用程序要求必须有一次且仅一次的语义，那么上述三个步骤都必须提供一次且仅一次的语义。每条数 据都得保证，只能接收一次、只能计算一次、只能推送一次。

**Spark Streaming中实心这些语义的步骤如下：**

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

## 三、接收数据的容错语义 ##

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

**2、基于Receiver的数据源**  
对于基于Receiver的数据源，容错语义依赖于失败的场景和Receiver类型。

*  **可靠的Receiver：** 这种Receiver会在接收到了数据，并且将数据复制之后，对数据源执行确认操作。如果 Receiver在数据接收和复制完成之前，就失败了，那么数据源对于缓存的数据会接收不到确认，此时，当 Receiver重启之后，数据源会重新发送数据，没有数据会丢失。
 *  **不可靠的Receiver：** 这种Receiver不会发送确认操作，因此当Worker或者Driver节点失败的时候，可能会导致 数据丢失。

不同的Receiver，提供了不同的语义。如果Worker节点失败了，那么使用的是可靠的Receiver 的话，没有数据会丢失。使用的是不可靠的Receiver的话，接收到，但是还没复制的数据，可能会 丢失。如果Driver节点失败的话，所有过去接收到的，和复制过缓存在内存中的数据，全部会丢失。

要避免这种过去接收的所有数据都丢失的问题，Spark从1.2版本开始，引入了预写日志机制， 可以将Receiver接收到的数据保存到容错存储中。如果使用可靠的Receiver，并且还开启了预写日 志机制，那么可以保证数据零丢失。这种情况下，会提供至少一次的保障。（Kafka是可以实现可靠 Receiver的）

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

## 四、输出数据的容错语义 ##

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

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

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

## 五、storm的容错语义 ##

Storm首先，它可以实现消息的高可靠性，就是说，它有一个机制，叫做Acker机制，可以保证， 如果消息处理失败，那么就重新发送。保证了，至少一次的容错语义。但是光靠这个，还是不行， 数据可能会重复。

Storm提供了非常非常完善的事务机制，可以实现一次且仅一次的事务机制。事务Topology、 透明的事务Topology、非透明的事务Topology，可以应用各种各样的情况。对实现一次且仅一次的 这种语义的支持，做的非常非常好。用事务机制，可以获得它内部提供的一个唯一的id，然后基于 这个id，就可以实现，output操作，输出，推送数据的时候，先判断，该数据是否更新过，如果没 有的话，就更新；如果更新过，就不要重复更新了。

所以，至少，在容错 / 事务机制方面，我觉得Spark Streaming还有很大的空间可以发展。特 别是对于output操作的一次且仅一次的语义支持！

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！