Spark-Streaming容错语义

一、背景

为了理解Spark Streaming提供的语义，我们先回顾西Spark RDD的基本容错语义学。

RDD是一个不可变的、确定性可重新计算的分布式数据集。每个RDD都记住在容错输入数据集上用于创建它的确定性操作的沿袭。
如果RDD的任何分区由于工作节点故障而丢失，则可以使用操作沿袭从原始容错数据集重新计算该分区。
假设所有RDD转换都是确定性的，最终转换后的RDD中的数据将始终相同，而不管Spark集群中的故障如何

Spark对HDFS或S3等容错文件系统中的数据进行操作。因此，从容错数据生成的所有RDD也是容错的。然而，Spark Streaming并非如此，因为在大多数情况下，数据是通过网络接收的（使用fileStream时除外）。为了实现所有生成的RDD的相同容错属性，接收到的数据将在集群中工作节点的多个Spark executors 之间复制（默认复制因子为2）。这导致系统中有两种数据需要在发生故障时恢复：

接收和复制的数据-此数据在单个工作节点发生故障时幸存下来，因为它的副本存在于其他节点之一上
已接收但为复制而缓冲的数据-由于未复制，因此恢复此数据的唯一方法是从源再次获取它

此外，我们应该关注两种失败：

工作节点的故障-任何运行执行器的工作节点都可能发生故障，并且这些节点上的所有内存数据都将丢失。如果任何接收器在故障节点上运行，那么它们的缓冲数据将丢失。
驱动程序节点的故障-如果运行Spark Streaming应用程序的驱动程序节点发生故障，那么显然SparkContext丢失了，并且所有具有内存数据的执行程序都丢失了。

有了这些基础知识，我们下面开始学习Spark Streaming的容错语义学

二、整体语义

流系统的语义学通常是根据系统可以处理每条记录的次数来捕获的。系统可以在所有可能的操作条件下（尽管有故障等）提供三种类型的保证。

最多处理一次：每条记录要么处理一次，要么根本不处理
至少一次：每条记录将被处理一次或多次。这比最多一次更强，因为它确保没有数据丢失。但可能会有重复。
精确一次：每条记录将被精确处理一次 - -- 没有数据丢失，也没有数据被多次处理。这显然是三者中最强大的保证。

在任何流处理系统中，广义上讲，处理数据有三个步骤。

接收数据：使用接收器或其他方式从源接收数据。
转换数据：使用DStream和RDD转换转换接收到的数据
推送数据：最终转换后的数据被推送到外部系统，如文件系统、数据库、仪表板等

如果一个流式应用程序必须实现端到端的精确一次保证，那么每个步骤都必须提供精确一次保证。也就是说，每条记录必须精确接收一次，精确转换一次，并精确推送到下游系统一次。Spark Streaming采用的是RDD来处理数据，RDD中间的转化操作都是迭代器模式，可以保证所有接收到的数据将只处理一次。即使出现故障，只要接收到的输入数据是可访问的，最终转换的RDD将始终具有相同的内容。这样就剩下接收数据和推送数据的保证，这两点我们再后面结合不同的输入源提供的保证以及下游系统的不同来进行详细分析。

三、接收数据语义

不同的输入源提供不同的保证，从至少一次到恰好一次。

1、输入源是文件

如果所有输入数据都已经存在于像HDFS这样的容错文件系统中，Spark Streaming总是可以从任何故障中恢复并处理所有数据。这给出了一次语义学，这意味着无论发生什么故障，所有数据都将被处理一次。

2、输入源是接收器（Receiver）

对于基于接收器的输入源，容错语义学取决于故障场景和接收器类型。正如我们之前讨论的，有两种类型的接收器：

可靠的接收器——这些接收器只有在确保接收到的数据已经被复制后才会确认可靠的来源。如果这样的接收器发生故障，源将不会收到缓冲（未复制）数据的确认。因此，如果接收器重新启动，源将重新发送数据，并且不会因故障而丢失数据。
不可靠的接收器-这种接收器不发送确认，因此当它们由于工作人员或驱动程序故障而失败时可能会丢失数据

根据使用的接收器类型，如果工作节点发生故障，那么可靠的接收器不会丢失数据。对于不可靠的接收器，接收但未复制的数据可能会丢失。如果driver 发生故障，那么除了这些丢失之外，所有过去在内存中接收和复制的数据都将丢失。这将影响有状态转换的结果。

为了避免过去接收到的数据丢失，Spark 1.2引入了预写日志，将接收到的数据保存到容错存储中。由于启用了预写日志和可靠的接收器，数据丢失为零。就语义学而言，它提供了至少一次保证。

因此推荐采用的模式为：带有预写日志的Spark 1.2或更高版本

3、输入源是Kafka的Direct API

在Spark 1.3中，引入了一个新的Kafka Direct API，它可以确保Spark Streaming只接收一次所有Kafka数据。

四、输出数据语义

输出操作（如foreachRDD）至少有一次语义学，也就是说，在worker 节点失败的情况下，转换后的数据可能会多次写入外部实体。虽然这对于使用saveAs***Files操作保存到文件系统是可以接受的（因为文件将被相同的数据覆盖），但可能需要额外的努力来实现一次语义学。有两种方法。

1、幂等更新：多次尝试总是写入相同的数据。例如，SaveAs***Files总是将相同的数据写入生成的文件。

2、事务性更新：所有更新都是以事务性方式进行的，因此更新仅以原子方式进行一次。

使用批处理时间（在foreachRDD中可用）和RDD的分区索引来创建标识符。此标识符唯一标识流应用程序中的blob数据
使用标识符以事务方式（即仅一次原子方式）使用此blob更新外部系统。也就是说，如果标识符尚未提交，请原子方式提交分区数据和标识符。否则，如果已经提交，请跳过更新。

dstream.foreachRDD { (rdd, time) =>
  rdd.foreachPartition { partitionIterator =>
    val partitionId = TaskContext.get.partitionId()
    val uniqueId = generateUniqueId(time.milliseconds, partitionId)
    // 使用此uniqueId在partitionIterator中事务性提交数据
  }
}