总之，元数据检查点主要用于从驱动程序故障中恢复，而数据或 RDD 检查点即使在基础功能中也是必要的（如果使用了有状态转换）。

何时启用检查点

对于具有以下任何要求的应用程序，必须启用检查点：

• 使用有状态转换 - 如果应用程序中使用了 updateStateByKey 或 reduceByKeyAndWindow（带有反函数），则必须提供检查点目录以允许定期进行 RDD 检查点。
• 从运行应用程序的驱动程序故障中恢复 - 元数据检查点用于通过进度信息进行恢复。

注意，没有上述有状态转换的简单流应用程序可以在不启用检查点的情况下运行。在这种情况下，从驱动程序故障中恢复也将是部分的（可能会丢失一些已接收但未处理的数据）。这通常是可以接受的，许多人以这种方式运行 Spark Streaming 应用程序。对非 Hadoop 环境的支持预计在未来会得到改善。

如何配置检查点

可以通过在容错、可靠的文件系统（例如 HDFS、S3 等）中设置一个目录来启用检查点，检查点信息将保存到该目录中。这是通过使用 streamingContext.checkpoint(checkpointDirectory) 完成的。这将允许你使用上述有状态转换。此外，如果你想让应用程序从驱动程序故障中恢复，你应该重写你的流应用程序以具有以下行为：

• 当程序第一次启动时，它将创建一个新的 StreamingContext，设置所有流，然后调用 start()。
• 当程序在故障后重新启动时，它将从检查点目录中的检查点数据重新创建一个 StreamingContext。

通过使用 StreamingContext.getOrCreate，这种行为变得简单。用法如下：

# Function to create and setup a new StreamingContext
def functionToCreateContext():
    sc = SparkContext(...)  # new context
    ssc = StreamingContext(...)
    lines = ssc.socketTextStream(...)  # create DStreams
    ...
    ssc.checkpoint(checkpointDirectory)  # set checkpoint directory
    return ssc

# Get StreamingContext from checkpoint data or create a new one
context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext)

# Do additional setup on context that needs to be done,
# irrespective of whether it is being started or restarted
context. ...

# Start the context
context.start()
context.awaitTermination()

如果 checkpointDirectory 存在，则上下文将从检查点数据重新创建。如果目录不存在（即第一次运行），则将调用函数 functionToCreateContext 来创建一个新上下文并设置 DStreams。查看 Python 示例 recoverable_network_wordcount.py。此示例将网络数据的单词计数附加到文件中。

你还可以通过使用 StreamingContext.getOrCreate(checkpointDirectory, None) 显式地从检查点数据创建 StreamingContext 并启动计算。

通过使用 StreamingContext.getOrCreate，这种行为变得简单。用法如下：

// Function to create and setup a new StreamingContext
def functionToCreateContext(): StreamingContext = {
  val ssc = new StreamingContext(...)   // new context
  val lines = ssc.socketTextStream(...) // create DStreams
  ...
  ssc.checkpoint(checkpointDirectory)   // set checkpoint directory
  ssc
}

// Get StreamingContext from checkpoint data or create a new one
val context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)

// Do additional setup on context that needs to be done,
// irrespective of whether it is being started or restarted
context. ...

// Start the context
context.start()
context.awaitTermination()

如果 checkpointDirectory 存在，则上下文将从检查点数据重新创建。如果目录不存在（即第一次运行），则将调用函数 functionToCreateContext 来创建一个新上下文并设置 DStreams。查看 Scala 示例 RecoverableNetworkWordCount。此示例将网络数据的单词计数附加到文件中。

通过使用 JavaStreamingContext.getOrCreate，这种行为变得简单。用法如下：

// Create a factory object that can create and setup a new JavaStreamingContext
JavaStreamingContextFactory contextFactory = new JavaStreamingContextFactory() {
  @Override public JavaStreamingContext create() {
    JavaStreamingContext jssc = new JavaStreamingContext(...);  // new context
    JavaDStream<String> lines = jssc.socketTextStream(...);     // create DStreams
    ...
    jssc.checkpoint(checkpointDirectory);                       // set checkpoint directory
    return jssc;
  }
};

// Get JavaStreamingContext from checkpoint data or create a new one
JavaStreamingContext context = JavaStreamingContext.getOrCreate(checkpointDirectory, contextFactory);

// Do additional setup on context that needs to be done,
// irrespective of whether it is being started or restarted
context. ...

// Start the context
context.start();
context.awaitTermination();

如果 checkpointDirectory 存在，则上下文将从检查点数据重新创建。如果目录不存在（即第一次运行），则将调用函数 contextFactory 来创建一个新上下文并设置 DStreams。查看 Java 示例 JavaRecoverableNetworkWordCount。此示例将网络数据的单词计数附加到文件中。

除了使用 getOrCreate 之外，还需要确保驱动程序进程在故障时自动重新启动。这只能由用于运行应用程序的部署基础结构来完成。这将在 部署 部分进一步讨论。

注意，RDD 的检查点会产生保存到可靠存储的成本。这可能会导致 RDD 被检查点的那些批次的处理时间增加。因此，需要仔细设置检查点间隔。在小批次大小（例如 1 秒）下，每批次检查点可能会显著降低操作吞吐量。相反，检查点太不频繁会导致血统和任务大小增加，这可能会产生不利影响。对于需要 RDD 检查点的有状态转换，默认间隔是批处理间隔的倍数，至少为 10 秒。它可以通过使用 dstream.checkpoint(checkpointInterval) 来设置。通常，5 到 10 个 DStream 滑动间隔的检查点间隔是一个很好的尝试设置。

---

累加器、广播变量和检查点

累加器 和 广播变量 无法从 Spark Streaming 中的检查点恢复。如果你启用了检查点并同时使用了 累加器 或 广播变量，则必须为 累加器 和 广播变量 创建惰性实例化的单例实例，以便它们可以在驱动程序在故障后重新启动时重新实例化。以下示例展示了这一点。

def getWordExcludeList(sparkContext):
    if ("wordExcludeList" not in globals()):
        globals()["wordExcludeList"] = sparkContext.broadcast(["a", "b", "c"])
    return globals()["wordExcludeList"]

def getDroppedWordsCounter(sparkContext):
    if ("droppedWordsCounter" not in globals()):
        globals()["droppedWordsCounter"] = sparkContext.accumulator(0)
    return globals()["droppedWordsCounter"]

def echo(time, rdd):
    # Get or register the excludeList Broadcast
    excludeList = getWordExcludeList(rdd.context)
    # Get or register the droppedWordsCounter Accumulator
    droppedWordsCounter = getDroppedWordsCounter(rdd.context)

    # Use excludeList to drop words and use droppedWordsCounter to count them
    def filterFunc(wordCount):
        if wordCount[0] in excludeList.value:
            droppedWordsCounter.add(wordCount[1])
            False
        else:
            True

    counts = "Counts at time %s %s" % (time, rdd.filter(filterFunc).collect())

wordCounts.foreachRDD(echo)

object WordExcludeList {

  @volatile private var instance: Broadcast[Seq[String]] = null

  def getInstance(sc: SparkContext): Broadcast[Seq[String]] = {
    if (instance == null) {
      synchronized {
        if (instance == null) {
          val wordExcludeList = Seq("a", "b", "c")
          instance = sc.broadcast(wordExcludeList)
        }
      }
    }
    instance
  }
}

object DroppedWordsCounter {

  @volatile private var instance: LongAccumulator = null

  def getInstance(sc: SparkContext): LongAccumulator = {
    if (instance == null) {
      synchronized {
        if (instance == null) {
          instance = sc.longAccumulator("DroppedWordsCounter")
        }
      }
    }
    instance
  }
}

wordCounts.foreachRDD { (rdd: RDD[(String, Int)], time: Time) =>
  // Get or register the excludeList Broadcast
  val excludeList = WordExcludeList.getInstance(rdd.sparkContext)
  // Get or register the droppedWordsCounter Accumulator
  val droppedWordsCounter = DroppedWordsCounter.getInstance(rdd.sparkContext)
  // Use excludeList to drop words and use droppedWordsCounter to count them
  val counts = rdd.filter { case (word, count) =>
    if (excludeList.value.contains(word)) {
      droppedWordsCounter.add(count)
      false
    } else {
      true
    }
  }.collect().mkString("[", ", ", "]")
  val output = "Counts at time " + time + " " + counts
})

class JavaWordExcludeList {

  private static volatile Broadcast<List<String>> instance = null;

  public static Broadcast<List<String>> getInstance(JavaSparkContext jsc) {
    if (instance == null) {
      synchronized (JavaWordExcludeList.class) {
        if (instance == null) {
          List<String> wordExcludeList = Arrays.asList("a", "b", "c");
          instance = jsc.broadcast(wordExcludeList);
        }
      }
    }
    return instance;
  }
}

class JavaDroppedWordsCounter {

  private static volatile LongAccumulator instance = null;

  public static LongAccumulator getInstance(JavaSparkContext jsc) {
    if (instance == null) {
      synchronized (JavaDroppedWordsCounter.class) {
        if (instance == null) {
          instance = jsc.sc().longAccumulator("DroppedWordsCounter");
        }
      }
    }
    return instance;
  }
}

wordCounts.foreachRDD((rdd, time) -> {
  // Get or register the excludeList Broadcast
  Broadcast<List<String>> excludeList = JavaWordExcludeList.getInstance(new JavaSparkContext(rdd.context()));
  // Get or register the droppedWordsCounter Accumulator
  LongAccumulator droppedWordsCounter = JavaDroppedWordsCounter.getInstance(new JavaSparkContext(rdd.context()));
  // Use excludeList to drop words and use droppedWordsCounter to count them
  String counts = rdd.filter(wordCount -> {
    if (excludeList.value().contains(wordCount._1())) {
      droppedWordsCounter.add(wordCount._2());
      return false;
    } else {
      return true;
    }
  }).collect().toString();
  String output = "Counts at time " + time + " " + counts;
}

部署应用程序

本节讨论部署 Spark Streaming 应用程序的步骤。

要求

要运行 Spark Streaming 应用程序，你需要具备以下条件：

• 具有集群管理器的集群 - 这是任何 Spark 应用程序的一般要求，并在 部署指南 中进行了详细讨论。

• 打包应用程序 JAR - 你必须将流应用程序编译为 JAR。如果你使用 spark-submit 启动应用程序，则无需在 JAR 中提供 Spark 和 Spark Streaming。但是，如果你的应用程序使用了 高级源（例如 Kafka），则必须将它们链接到的额外构件及其依赖项打包在用于部署应用程序的 JAR 中。例如，使用 KafkaUtils 的应用程序必须在其应用程序 JAR 中包含 spark-streaming-kafka-0-10_2.13 及其所有传递依赖项。

• 为执行器配置足够的内存 - 由于接收到的数据必须存储在内存中，因此必须将执行器配置为具有足够的内存来保存接收到的数据。注意，如果你正在执行 10 分钟的窗口操作，系统必须在内存中至少保留过去 10 分钟的数据。因此，应用程序的内存需求取决于其中使用的操作。

• 配置检查点 - 如果流应用程序需要，则必须将 Hadoop API 兼容的容错存储（例如 HDFS、S3 等）中的目录配置为检查点目录，并以可以使用检查点信息进行故障恢复的方式编写流应用程序。有关详细信息，请参阅 检查点 部分。

• 配置应用程序驱动程序的自动重启 - 为了从驱动程序故障中自动恢复，用于运行流应用程序的部署基础结构必须监控驱动程序进程，并在驱动程序故障时重新启动它。不同的 集群管理器 有不同的工具来实现这一点。
• Spark Standalone - 可以提交 Spark 应用程序驱动程序以在 Spark Standalone 集群中运行（参阅 集群部署模式），也就是说，应用程序驱动程序本身运行在其中一个工作节点上。此外，可以指示 Standalone 集群管理器 监督 驱动程序，并在驱动程序由于非零退出代码或运行驱动程序的节点故障而失败时重新启动它。有关详细信息，请参阅 Spark Standalone 指南 中的 集群模式 和 监督。
• YARN - Yarn 支持类似的自动重启应用程序的机制。请参阅 YARN 文档以了解更多详细信息。

配置预写日志 (Write-ahead Logs) - 自 Spark 1.2 起，我们引入了 预写日志 以实现强大的容错保证。如果启用，从接收器接收到的所有数据都会写入配置检查点目录中的预写日志中。这可以防止驱动程序恢复时的数据丢失，从而确保零数据丢失（在 容错语义 部分详细讨论）。可以通过将 配置参数 spark.streaming.receiver.writeAheadLog.enable 设置为 true 来启用此功能。但是，这些更强的语义可能会以牺牲单个接收器的接收吞吐量为代价。这可以通过运行 更多并行接收器 来增加总吞吐量来纠正。此外，建议在启用预写日志时禁用 Spark 内接收数据的复制，因为日志已存储在复制存储系统中。这可以通过将输入流的存储级别设置为 StorageLevel.MEMORY_AND_DISK_SER 来完成。在使用 S3（或任何不支持刷新操作的文件系统）进行 预写日志 时，请记住启用 spark.streaming.driver.writeAheadLog.closeFileAfterWrite 和 spark.streaming.receiver.writeAheadLog.closeFileAfterWrite。有关更多详细信息，请参阅 Spark Streaming 配置。注意，当启用 I/O 加密时，Spark 不会加密写入预写日志的数据。如果需要预写日志数据的加密，它应存储在原生支持加密的文件系统中。

升级应用程序代码

如果需要使用新的应用程序代码升级正在运行的 Spark Streaming 应用程序，则有两种可能的机制。

• 升级后的 Spark Streaming 应用程序与现有应用程序并行启动和运行。一旦新的应用程序（接收与旧应用程序相同的数据）预热并准备就绪，旧的应用程序就可以关闭。注意，这对于支持将数据发送到两个目的地（即之前和升级后的应用程序）的数据源来说是可以做到的。

• 现有应用程序被优雅地关闭（有关优雅关闭选项，请参阅 StreamingContext.stop(...) 或 JavaStreamingContext.stop(...)），这确保了在关闭之前已接收的数据得到完全处理。然后可以启动升级后的应用程序，它将从前一个应用程序停止的地方开始处理。注意，这只能在支持源端缓冲（如 Kafka）的输入源中完成，因为在之前应用程序关闭而升级后的应用程序尚未启动时，数据需要被缓冲。并且无法从预升级代码的先前检查点信息重新启动。检查点信息本质上包含序列化的 Scala/Java/Python 对象，尝试使用新的、修改后的类反序列化对象可能会导致错误。在这种情况下，要么使用不同的检查点目录启动升级后的应用程序，要么删除先前的检查点目录。

---

监控应用程序

除了 Spark 的 监控功能 之外，还有特定于 Spark Streaming 的附加功能。当使用 StreamingContext 时，Spark Web UI 会显示一个额外的 Streaming 选项卡，其中显示有关运行中接收器（接收器是否处于活动状态、接收到的记录数、接收器错误等）和已完成批次（批处理时间、排队延迟等）的统计信息。这可用于监控流应用程序的进度。

Web UI 中的以下两个指标尤为重要：

• 处理时间 (Processing Time) - 处理每个数据批次所需的时间。
• 调度延迟 (Scheduling Delay) - 批次在队列中等待之前批次处理完成的时间。

如果批处理时间持续超过批处理间隔和/或排队延迟不断增加，则表明系统无法以生成数据的速度处理批次，并且正在落后。在这种情况下，请考虑 减少 批处理时间。

Spark Streaming 程序的进度也可以使用 StreamingListener 接口进行监控，该接口允许你获取接收器状态和处理时间。注意，这是一个开发人员 API，预计在未来会得到改进（即报告更多信息）。

性能调优

在集群上获得 Spark Streaming 应用程序的最佳性能需要进行一些调整。本节解释了可以调整以提高应用程序性能的多个参数和配置。在宏观层面上，你需要考虑两件事：

1. 通过有效地使用集群资源来减少每个数据批次的处理时间。

2. 设置正确的批处理大小，以便数据批次可以以接收它们的速度进行处理（即数据处理跟上数据摄入）。

减少批处理时间

Spark 中可以进行许多优化以最大限度地减少每个批次的处理时间。这些已在 调优指南 中进行了详细讨论。本节重点介绍了其中一些最重要的优化。

数据接收的并行度

通过网络接收数据（如 Kafka、套接字等）要求数据被反序列化并存储在 Spark 中。如果数据接收成为系统中的瓶颈，则考虑并行化数据接收。注意，每个输入 DStream 创建一个单一的接收器（运行在 worker 机器上），该接收器接收单个数据流。因此，接收多个数据流可以通过创建多个输入 DStreams 并配置它们从源接收数据流的不同分区来实现。例如，接收两个主题数据的单个 Kafka 输入 DStream 可以拆分为两个 Kafka 输入流，每个流仅接收一个主题。这将运行两个接收器，从而允许并行接收数据，进而增加总吞吐量。可以将这些多个 DStreams 连接在一起以创建一个单一的 DStream。然后，可以对统一后的流应用之前应用于单个输入 DStream 的转换。这是按照以下方式完成的：

numStreams = 5
kafkaStreams = [KafkaUtils.createStream(...) for _ in range (numStreams)]
unifiedStream = streamingContext.union(*kafkaStreams)
unifiedStream.pprint()

val numStreams = 5
val kafkaStreams = (1 to numStreams).map { i => KafkaUtils.createStream(...) }
val unifiedStream = streamingContext.union(kafkaStreams)
unifiedStream.print()

int numStreams = 5;
List<JavaPairDStream<String, String>> kafkaStreams = new ArrayList<>(numStreams);
for (int i = 0; i < numStreams; i++) {
  kafkaStreams.add(KafkaUtils.createStream(...));
}
JavaPairDStream<String, String> unifiedStream = streamingContext.union(kafkaStreams.get(0), kafkaStreams.subList(1, kafkaStreams.size()));
unifiedStream.print();

另一个应该考虑的参数是接收器的块间隔 (block interval)，该间隔由 配置参数 spark.streaming.blockInterval 决定。对于大多数接收器，接收到的数据在存储到 Spark 内存之前会被合并成数据块。每批次中的块数决定了在 map 类转换中用于处理接收到的数据的任务数。每批次每个接收器的任务数将大约为（批处理间隔 / 块间隔）。例如，200 毫秒的块间隔将为每 2 秒的批次创建 10 个任务。如果任务数太少（即少于每台机器的核心数），那么它将是低效的，因为所有可用核心都不会被用于处理数据。要增加给定批处理间隔的任务数，请减小块间隔。但是，块间隔的建议最小值为约 50 毫秒，低于此值可能会出现任务启动开销问题。

接收具有多个输入流/接收器的数据的替代方法是显式重分区输入数据流（使用 inputStream.repartition(<number of partitions>)）。这会在进一步处理之前将接收到的数据批次分发到集群中指定数量的机器上。

对于直接流，请参考 Spark Streaming + Kafka 集成指南

数据处理的并行度

如果计算的任何阶段所使用的并行任务数量不够高，集群资源可能会被利用不足。例如，对于像 reduceByKey 和 reduceByKeyAndWindow 这样的分布式 reduce 操作，默认的并行任务数量由 spark.default.parallelism 配置属性控制。你可以将并行度作为参数传递（参见 PairDStreamFunctions 文档），或者设置 spark.default.parallelism 配置属性来更改默认值。

数据序列化

通过调整序列化格式可以降低数据序列化的开销。在流处理中，有两种类型的数据需要序列化。

• 输入数据：默认情况下，通过接收器（Receivers）接收的输入数据以 StorageLevel.MEMORY_AND_DISK_SER_2 存储在执行器（executors）的内存中。这意味着，为了减少 GC 开销，数据会被序列化为字节，并进行复制以容忍执行器故障。此外，数据首先保留在内存中，只有在内存不足以容纳流计算所需的所有输入数据时，才会溢出到磁盘。这种序列化显然存在开销——接收器必须反序列化接收到的数据，并使用 Spark 的序列化格式重新序列化它。

• 流式操作生成的持久化 RDD：流式计算生成的 RDD 可能会持久化在内存中。例如，窗口操作会将数据持久化在内存中，因为它们会被多次处理。然而，与 Spark Core 默认的 StorageLevel.MEMORY_ONLY 不同，流式计算生成的持久化 RDD 默认使用 StorageLevel.MEMORY_ONLY_SER（即序列化）进行持久化，以最小化 GC 开销。

在这两种情况下，使用 Kryo 序列化都可以减少 CPU 和内存开销。有关更多详细信息，请参阅 Spark 调优指南。对于 Kryo，请考虑注册自定义类，并禁用对象引用跟踪（参见配置指南中与 Kryo 相关的配置）。

在流式应用所需保留的数据量不大的特定情况下，将数据（上述两种类型）作为反序列化对象进行持久化，而不产生过多的 GC 开销可能是可行的。例如，如果你使用的批处理间隔为几秒钟且没有窗口操作，那么你可以尝试通过相应地显式设置存储级别来禁用持久化数据中的序列化。这将减少由于序列化导致的 CPU 开销，可能在不产生过多 GC 开销的情况下提高性能。

设置合理的批处理间隔

为了使运行在集群上的 Spark Streaming 应用保持稳定，系统应该能够以与接收数据相同的速度处理数据。换句话说，批次数据的处理速度应与生成速度一致。可以通过在流处理 Web UI 中监控处理时间来判断应用是否满足这一点，其中批处理时间应小于批处理间隔。

根据流计算的性质，所使用的批处理间隔可能会对应用在固定集群资源下所能维持的数据速率产生重大影响。例如，让我们考虑早期的 WordCountNetwork 示例。对于特定的数据速率，系统可能能够跟上每 2 秒报告一次词频（即 2 秒的批处理间隔），但无法跟上每 500 毫秒一次。因此，批处理间隔的设置需要确保生产环境中的预期数据速率能够被维持。

确定应用正确批处理大小的一个好方法是使用保守的批处理间隔（例如 5-10 秒）和低数据速率进行测试。为了验证系统是否能够跟上数据速率，你可以检查每个已处理批次的端到端延迟值（查看 Spark 驱动程序的 log4j 日志中的“Total delay”，或使用 StreamingListener 接口）。如果延迟保持在与批处理大小相当的水平，则系统是稳定的。否则，如果延迟持续增加，则意味着系统无法跟上，因此是不稳定的。一旦你对稳定配置有了概念，就可以尝试提高数据速率和/或减小批处理大小。注意，由于临时数据速率增加导致的短暂延迟增加是可以接受的，只要延迟回落到较低值（即小于批处理大小）即可。

内存调优

关于调整 Spark 应用的内存使用和 GC 行为，调优指南中已有详细讨论。强烈建议阅读该指南。在本节中，我们将专门讨论 Spark Streaming 应用背景下的一些调优参数。

Spark Streaming 应用所需的集群内存量在很大程度上取决于所使用的转换类型。例如，如果你想对过去 10 分钟的数据使用窗口操作，那么你的集群应有足够的内存来容纳 10 分钟的数据。或者，如果你想对大量键使用 updateStateByKey，那么所需的内存会很高。相反，如果你只是进行简单的 map-filter-store 操作，那么所需的内存则较低。

通常情况下，由于通过接收器接收的数据以 StorageLevel.MEMORY_AND_DISK_SER_2 存储，无法放入内存的数据会溢出到磁盘。这可能会降低流应用的处理性能，因此建议根据流应用的需求提供充足的内存。最好在小规模环境下观察内存使用情况，并进行相应估算。

内存调优的另一个方面是垃圾回收（GC）。对于需要低延迟的流应用，由 JVM 垃圾回收引起的大停顿是不可取的。

有几个参数可以帮助你调整内存使用和 GC 开销。

• DStream 的持久化级别：如前文在 数据序列化 部分所述，输入数据和 RDD 默认以序列化字节形式持久化。与反序列化持久化相比，这减少了内存使用和 GC 开销。启用 Kryo 序列化进一步减小了序列化大小和内存使用。通过压缩（参见 Spark 配置 spark.rdd.compress）可以以 CPU 时间为代价进一步减少内存使用。

• 清理旧数据：默认情况下，DStream 转换生成的所有输入数据和持久化 RDD 都会被自动清理。Spark Streaming 根据所使用的转换决定何时清理数据。例如，如果你使用 10 分钟的窗口操作，那么 Spark Streaming 将保留最近 10 分钟的数据，并主动丢弃更早的数据。通过设置 streamingContext.remember，可以将数据保留更长的时间（例如为了交互式查询旧数据）。

• 其他技巧：为了进一步降低 GC 开销，这里还有一些建议可供尝试。

  • 使用 OFF_HEAP 存储级别持久化 RDD。详细信息请参阅 Spark 编程指南。
  • 使用更多具有较小堆内存的执行器（executors）。这将减少每个 JVM 堆内的 GC 压力。

需要记住的要点

• 一个 DStream 与单个接收器关联。为了实现读取并行性，需要创建多个接收器，即多个 DStream。接收器在执行器内运行，占用一个核心。在预定接收器插槽后，请确保有足够的内核用于处理，即 spark.cores.max 应将接收器插槽考虑在内。接收器以轮询方式分配给执行器。

• 当数据从流源接收时，接收器会创建数据块。每隔 blockInterval 毫秒会生成一个新的数据块。在 batchInterval 期间创建 N 个数据块，其中 N = batchInterval/blockInterval。这些块由当前执行器的 BlockManager 分发到其他执行器的 BlockManager。之后，运行在驱动程序上的 Network Input Tracker 会获知块的位置以进行进一步处理。

• 在 batchInterval 期间创建的块会在驱动程序上创建一个 RDD。batchInterval 期间生成的块是 RDD 的分区。每个分区都是 Spark 中的一个任务。blockInterval == batchInterval 意味着只创建一个分区，并且通常会在本地处理。

• 无论 blockInterval 如何，块上的 map 任务都会在拥有这些块的执行器（接收块的执行器和复制块的执行器）上处理，除非触发了非本地调度。更大的 blockInterval 意味着更大的块。较高的 spark.locality.wait 值增加了在本地节点处理块的机会。需要在这两个参数之间找到平衡，以确保更大的块能够被本地处理。

• 你可以通过调用 inputDstream.repartition(n) 来定义分区数量，而不是依赖 batchInterval 和 blockInterval。这会随机重排 RDD 中的数据以创建 n 个分区。是的，为了获得更高的并行度。虽然这会以 shuffle 为代价。RDD 的处理由驱动程序的任务调度器作为作业进行调度。在给定时间点，只有一个作业处于活动状态。因此，如果一个作业正在执行，其他作业将被排队。

• 如果你有两个 DStream，将会形成两个 RDD，并创建两个作业，它们将依次调度。为了避免这种情况，你可以合并两个 DStream。这将确保为两个 DStream 的 RDD 形成一个单一的 unionRDD。此 unionRDD 然后被视为单个作业。但是，RDD 的分区不受此影响。

• 如果批处理时间超过了 batchInterval，那么显然接收器的内存将开始填满，最终会抛出异常（很可能是 BlockNotFoundException）。目前无法暂停接收器。可以使用 SparkConf 配置 spark.streaming.receiver.maxRate 来限制接收器的速率。

容错语义

在本节中，我们将讨论 Spark Streaming 应用在发生故障时的行为。

背景

要理解 Spark Streaming 提供的语义，让我们回顾一下 Spark RDD 的基本容错语义。

1. RDD 是一个不可变的、确定性可重计算的分布式数据集。每个 RDD 都记得用于在容错输入数据集上创建它的一系列确定性操作。
2. 如果 RDD 的任何分区由于工作节点故障而丢失，则可以使用操作的血缘关系从原始容错数据集重新计算该分区。
3. 假设所有的 RDD 转换都是确定的，无论 Spark 集群中发生什么故障，最终转换后的 RDD 中的数据将始终相同。

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

1. 接收并复制的数据 - 此数据可以抵御单个工作节点的故障，因为它的一份副本存在于其他节点上。
2. 已接收但缓冲用于复制的数据 - 由于此数据未被复制，恢复此数据的唯一方法是从源头重新获取它。

此外，我们应该关注两种类型的故障：

1. 工作节点故障 - 运行执行器的任何工作节点都可能发生故障，该节点上的所有内存数据都将丢失。如果故障节点上运行了任何接收器，则其缓冲的数据将会丢失。
2. 驱动节点故障 - 如果运行 Spark Streaming 应用的驱动节点发生故障，显然 SparkContext 会丢失，所有带有内存数据的执行器也会丢失。

有了这些基本知识，让我们来了解 Spark Streaming 的容错语义。

定义

流系统的语义通常通过每条记录被系统处理的次数来捕获。系统在所有可能的运行条件下（尽管有故障等）可以提供三种类型的保证：

1. 最多一次（At most once）：每条记录要么被处理一次，要么根本不处理。
2. 至少一次（At least once）：每条记录将被处理一次或多次。这比最多一次更强，因为它确保数据不会丢失。但可能会出现重复。
3. 恰好一次（Exactly once）：每条记录将被恰好处理一次 - 数据不会丢失，也不会被处理多次。这显然是三种保证中最强的一种。

基本语义

在任何流处理系统中，概括地说，处理数据有三个步骤。

1. 接收数据：数据通过接收器或其他方式从源头接收。

2. 转换数据：接收到的数据使用 DStream 和 RDD 转换进行转换。

3. 推送数据：最终转换后的数据被推送到文件系统、数据库、仪表板等外部系统。

如果流应用要实现端到端的恰好一次保证，那么每个步骤都必须提供恰好一次保证。也就是说，每条记录必须被恰好接收一次、恰好转换一次，并被恰好推送至下游系统一次。让我们在 Spark Streaming 的背景下了解这些步骤的语义。

1. 接收数据：不同的输入源提供不同的保证。这将在下一小节中详细讨论。

2. 转换数据：得益于 RDD 提供的保证，所有已接收的数据都将被恰好一次处理。即使发生故障，只要接收到的输入数据可访问，最终转换后的 RDD 内容将始终相同。

3. 推送数据：输出操作默认确保至少一次语义，因为它取决于输出操作的类型（是否幂等）以及下游系统的语义（是否支持事务）。但用户可以实现自己的事务机制来达到恰好一次语义。本节稍后将详细讨论。

已接收数据的语义

不同的输入源提供不同的保证，范围从至少一次到恰好一次。阅读以了解更多详情。

对于文件

如果所有输入数据都已经存在于 HDFS 等容错文件系统中，Spark Streaming 总能从任何故障中恢复并处理所有数据。这提供了恰好一次语义，意味着无论发生什么故障，所有数据都将被恰好处理一次。

对于基于接收器的源

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

1. 可靠接收器（Reliable Receiver） - 这些接收器仅在确保接收到的数据已被复制后才确认可靠源。如果此类接收器发生故障，源将不会收到针对缓冲（未复制）数据的确认。因此，如果接收器重启，源将重新发送数据，且不会因故障而丢失任何数据。
2. 不可靠接收器（Unreliable Receiver） - 此类接收器不发送确认，因此在因工作节点或驱动故障而失败时可能丢失数据。

根据所使用的接收器类型，我们实现以下语义。如果工作节点发生故障，可靠接收器不会丢失数据。使用不可靠接收器时，已接收但未复制的数据可能会丢失。如果驱动节点发生故障，除了上述损失外，过去所有在内存中接收并复制的数据都将丢失。这将影响有状态转换的结果。

为了避免过去接收到的数据丢失，Spark 1.2 引入了预写日志（Write Ahead Logs），它将接收到的数据保存到容错存储中。在启用预写日志且使用可靠接收器的情况下，数据不会丢失。就语义而言，它提供了至少一次保证。

下表总结了故障情况下的语义

使用 Kafka Direct API

在 Spark 1.3 中，我们引入了新的 Kafka Direct API，它可以确保所有 Kafka 数据都能被 Spark Streaming 恰好一次地接收。在此基础上，如果你实现了恰好一次的输出操作，就可以实现端到端的恰好一次保证。这种方法在 Kafka 集成指南中作了进一步讨论。

输出操作的语义

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

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

• 事务性更新：所有更新都以事务方式进行，确保更新原子地发生且仅发生一次。实现此目的的一种方法如下：

  • 使用批处理时间（在 foreachRDD 中可用）和 RDD 的分区索引创建一个标识符。此标识符唯一地标识流应用中的一个数据块（blob）。

  • 使用该标识符将数据块事务性地更新到外部系统（即恰好一次，原子地）。也就是说，如果标识符尚未提交，则原子地提交分区数据和标识符。否则，如果已经提交，则跳过更新。

    dstream.foreachRDD { (rdd, time) =>
      rdd.foreachPartition { partitionIterator =>
        val partitionId = TaskContext.get.partitionId()
        val uniqueId = generateUniqueId(time.milliseconds, partitionId)
        // use this uniqueId to transactionally commit the data in partitionIterator
      }
    }
    

后续步骤

• 其他指南
  • Kafka 集成指南
  • Kinesis 集成指南
  • 自定义接收器指南
• 第三方 DStream 数据源可以在 第三方项目中找到。
• API 文档
  • Python 文档
    • StreamingContext 和 DStream
  • Scala 文档
    • StreamingContext 和 DStream
    • KafkaUtils, KinesisUtils
  • Java 文档
    • JavaStreamingContext, JavaDStream 和 JavaPairDStream
    • KafkaUtils, KinesisUtils
• 更多示例请见 Python、Scala 和 Java
• 描述 Spark Streaming 的论文和视频。