Spark 流计算编程指南

• 注意
• 概述
• 快速示例
• 基本概念
  • 链接
  • 初始化 StreamingContext
  • 离散化流 (DStreams)
  • 输入 DStreams 和接收器
  • DStreams 上的转换
  • DStreams 上的输出操作
  • DataFrame 和 SQL 操作
  • MLlib 操作
  • 缓存/持久化
  • 检查点
  • 累加器、广播变量和检查点
  • 部署应用程序
  • 监控应用程序
• 性能调优
  • 减少批处理时间
  • 设置正确的批处理间隔
  • 内存调优
• 容错语义
• 下一步

注意

Spark Streaming 是 Spark 流计算引擎的上一代版本。Spark Streaming 不再更新，它是一个遗留项目。Spark 中有一个更新且更易于使用的流计算引擎，称为结构化流计算。您应该在流计算应用程序和管道中使用 Spark 结构化流计算。请参阅结构化流计算编程指南。

概述

Spark Streaming 是核心 Spark API 的扩展，支持对实时数据流进行可扩展、高吞吐量、容错的流处理。数据可以从 Kafka、Kinesis 或 TCP 套接字等多种来源获取，并且可以使用使用 map、reduce、join 和 window 等高级函数表达的复杂算法进行处理。最后，处理后的数据可以推送到文件系统、数据库和实时仪表板。实际上，您可以将 Spark 的机器学习和图计算算法应用于数据流。

在内部，它的工作原理如下。Spark Streaming 接收实时输入数据流，并将数据分成批次，然后由 Spark 引擎处理这些批次以生成最终的结果流。

Spark Streaming 提供了一个称为*离散化流*或*DStream*的高级抽象，它表示连续的数据流。DStreams 可以从 Kafka 和 Kinesis 等来源的输入数据流创建，也可以通过对其他 DStreams 应用高级操作来创建。在内部，DStream 表示为一系列RDD。

本指南介绍如何使用 DStreams 开始编写 Spark Streaming 程序。您可以使用 Scala、Java 或 Python（在 Spark 1.2 中引入）编写 Spark Streaming 程序，所有这些都在本指南中进行了介绍。您将在本指南中找到选项卡，让您可以在不同语言的代码片段之间进行选择。

**注意：**有一些 API 在 Python 中有所不同或不可用。在本指南中，您将找到标记 Python API，突出显示这些差异。

快速示例

在我们详细介绍如何编写自己的 Spark Streaming 程序之前，让我们快速了解一下一个简单的 Spark Streaming 程序是什么样的。假设我们要统计从监听 TCP 套接字的数据服务器接收到的文本数据中的单词数量。您只需执行以下操作。

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

# Create a local StreamingContext with two working thread and batch interval of 1 second
sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, 1)

# Create a DStream that will connect to hostname:port, like localhost:9999
lines = ssc.socketTextStream("localhost", 9999)

# Split each line into words
words = lines.flatMap(lambda line: line.split(" "))

# Count each word in each batch
pairs = words.map(lambda word: (word, 1))
wordCounts = pairs.reduceByKey(lambda x, y: x + y)

# Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.pprint()

ssc.start()             # Start the computation
ssc.awaitTermination()  # Wait for the computation to terminate

import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3

// Create a local StreamingContext with two working thread and batch interval of 1 second.
// The master requires 2 cores to prevent a starvation scenario.

val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
val ssc = new StreamingContext(conf, Seconds(1))

// Create a DStream that will connect to hostname:port, like localhost:9999
val lines = ssc.socketTextStream("localhost", 9999)

// Split each line into words
val words = lines.flatMap(_.split(" "))

import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
// Count each word in each batch
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)

// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print()

ssc.start()             // Start the computation
ssc.awaitTermination()  // Wait for the computation to terminate

import org.apache.spark.*;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.*;
import org.apache.spark.streaming.api.java.*;
import scala.Tuple2;

// Create a local StreamingContext with two working thread and batch interval of 1 second
SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));

// Create a DStream that will connect to hostname:port, like localhost:9999
JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);

// Split each line into words
JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());

// Count each word in each batch
JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));
JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey((i1, i2) -> i1 + i2);

// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print();

jssc.start();              // Start the computation
jssc.awaitTermination();   // Wait for the computation to terminate

如果您已经下载并构建了 Spark，则可以按如下方式运行此示例。您首先需要使用以下命令将 Netcat（在大多数类 Unix 系统中都能找到的小型实用程序）作为数据服务器运行

$ nc -lk 9999

然后，在另一个终端中，您可以使用以下命令启动示例

$ ./bin/spark-submit examples/src/main/python/streaming/network_wordcount.py localhost 9999

$ ./bin/run-example streaming.NetworkWordCount localhost 9999

$ ./bin/run-example streaming.JavaNetworkWordCount localhost 9999

然后，在运行 netcat 服务器的终端中键入的任何行都将被统计并在屏幕上每秒打印一次。它看起来像下面这样。

# TERMINAL 1:
# Running Netcat

$ nc -lk 9999

hello world



...

# TERMINAL 2: RUNNING network_wordcount.py

$ ./bin/spark-submit examples/src/main/python/streaming/network_wordcount.py localhost 9999
...
-------------------------------------------
Time: 2014-10-14 15:25:21
-------------------------------------------
(hello,1)
(world,1)
...

# TERMINAL 2: RUNNING NetworkWordCount

$ ./bin/run-example streaming.NetworkWordCount localhost 9999
...
-------------------------------------------
Time: 1357008430000 ms
-------------------------------------------
(hello,1)
(world,1)
...

# TERMINAL 2: RUNNING JavaNetworkWordCount

$ ./bin/run-example streaming.JavaNetworkWordCount localhost 9999
...
-------------------------------------------
Time: 1357008430000 ms
-------------------------------------------
(hello,1)
(world,1)
...

基本概念

接下来，我们将超越简单的示例，详细阐述 Spark Streaming 的基础知识。

链接

与 Spark 类似，Spark Streaming 可通过 Maven Central 获取。要编写您自己的 Spark Streaming 程序，您必须将以下依赖项添加到您的 SBT 或 Maven 项目中。

<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-streaming_2.12</artifactId>
    <version>3.5.1</version>
    <scope>provided</scope>
</dependency>

libraryDependencies += "org.apache.spark" % "spark-streaming_2.12" % "3.5.1" % "provided"

对于从 Spark Streaming 核心 API 中不存在的源（如 Kafka 和 Kinesis）提取数据，您必须将相应的工件 spark-streaming-xyz_2.12 添加到依赖项中。例如，一些常见的工件如下所示。

有关最新列表，请参阅 Maven 存储库，以获取受支持的源和工件的完整列表。

初始化 StreamingContext

要初始化 Spark Streaming 程序，必须创建一个 StreamingContext 对象，它是所有 Spark Streaming 功能的主要入口点。

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

sc = SparkContext(master, appName)
ssc = StreamingContext(sc, 1)

import org.apache.spark._
import org.apache.spark.streaming._

val conf = new SparkConf().setAppName(appName).setMaster(master)
val ssc = new StreamingContext(conf, Seconds(1))

import org.apache.spark.streaming._

val sc = ...                // existing SparkContext
val ssc = new StreamingContext(sc, Seconds(1))

import org.apache.spark.*;
import org.apache.spark.streaming.api.java.*;

SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
JavaStreamingContext ssc = new JavaStreamingContext(conf, new Duration(1000));

import org.apache.spark.streaming.api.java.*;

JavaSparkContext sc = ...   //existing JavaSparkContext
JavaStreamingContext ssc = new JavaStreamingContext(sc, Durations.seconds(1));

定义上下文后，您必须执行以下操作。

1. 通过创建输入 DStream 来定义输入源。
2. 通过对 DStream 应用转换和输出操作来定义流计算。
3. 使用 streamingContext.start() 开始接收数据并对其进行处理。
4. 使用 streamingContext.awaitTermination() 等待处理停止（手动或由于任何错误）。
5. 可以使用 streamingContext.stop() 手动停止处理。

需要记住的几点

• 一旦上下文启动，就不能设置或添加新的流计算。
• 一旦上下文停止，就不能重新启动。
• 在 JVM 中，一次只能有一个 StreamingContext处于活动状态。
• StreamingContext 上的 stop() 也会停止 SparkContext。要仅停止 StreamingContext，请将 stop() 的可选参数 stopSparkContext 设置为 false。
• 只要在上一个 StreamingContext 停止（不停止 SparkContext）之后创建下一个 StreamingContext，就可以重复使用 SparkContext 创建多个 StreamingContext。

离散化流 (DStreams)

离散流或 DStream 是 Spark Streaming 提供的基本抽象。它表示连续的数据流，可以是从源接收的输入数据流，也可以是通过转换输入数据流生成的已处理数据流。在内部，DStream 由一系列连续的 RDD 表示，RDD 是 Spark 对不可变分布式数据集的抽象（有关更多详细信息，请参阅Spark 编程指南）。DStream 中的每个 RDD 都包含来自特定时间间隔的数据，如下图所示。

应用于 DStream 的任何操作都会转换为对底层 RDD 的操作。例如，在将行流转换为单词的前面的示例中，flatMap 操作应用于 lines DStream 中的每个 RDD，以生成 words DStream 的 RDD。如下图所示。

这些底层 RDD 转换由 Spark 引擎计算。DStream 操作隐藏了大部分这些细节，并为开发人员提供了更高级别的 API 以方便使用。这些操作将在后面的部分中详细讨论。

输入 DStreams 和接收器

输入 DStream 是表示从流源接收的输入数据流的 DStream。在快速示例中，lines 是一个输入 DStream，因为它表示从 netcat 服务器接收的数据流。每个输入 DStream（文件流除外，本节稍后将讨论）都与一个 接收器（Scala 文档，Java 文档）对象相关联，该对象从源接收数据并将其存储在 Spark 的内存中以进行处理。

Spark Streaming 提供两类内置流源。

• 基本源：StreamingContext API 中直接可用的源。例如：文件系统和套接字连接。
• 高级源：Kafka、Kinesis 等源可通过额外的实用程序类获得。这些需要链接到额外的依赖项，如链接部分所述。

我们将在本节稍后讨论每个类别中存在的一些来源。

请注意，如果要在流应用程序中并行接收多个数据流，则可以创建多个输入 DStream（在性能调优部分中进一步讨论）。这将创建多个接收器，它们将同时接收多个数据流。但请注意，Spark 工作器/执行器是一个长时间运行的任务，因此它会占用分配给 Spark Streaming 应用程序的一个核心。因此，重要的是要记住，Spark Streaming 应用程序需要分配足够的内核（如果在本地运行，则为线程）来处理接收到的数据，以及运行接收器。

需要记住的几点

• 在本地运行 Spark Streaming 程序时，请勿使用“local”或“local[1]”作为主 URL。其中任何一个都意味着只有一个线程将用于在本地运行任务。如果您使用的是基于接收器的输入 DStream（例如套接字、Kafka 等），则单个线程将用于运行接收器，而没有线程用于处理接收到的数据。因此，在本地运行时，始终使用“local[n]”作为主 URL，其中 n > 要运行的接收器数量（有关如何设置主服务器的信息，请参阅Spark 属性）。

• 将逻辑扩展到在集群上运行，分配给 Spark Streaming 应用程序的核心数量必须大于接收器的数量。否则，系统将接收数据，但无法处理它。

基本来源

我们已经在快速示例中查看了 ssc.socketTextStream(...)，它从通过 TCP 套接字连接接收的文本数据创建 DStream。除了套接字之外，StreamingContext API 还提供了用于从文件创建 DStream 作为输入源的方法。

文件流

要从与 HDFS API 兼容的任何文件系统（即 HDFS、S3、NFS 等）上的文件读取数据，可以通过 StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass] 创建 DStream。

文件流不需要运行接收器，因此无需分配任何核心来接收文件数据。

对于简单的文本文件，最简单的方法是 StreamingContext.textFileStream(dataDirectory)。

streamingContext.textFileStream(dataDirectory)

streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)

streamingContext.textFileStream(dataDirectory)

streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);

streamingContext.textFileStream(dataDirectory);

如何监控目录

Spark Streaming 将监控目录 dataDirectory 并处理在该目录中创建的任何文件。

• 可以监控一个简单的目录，例如 "hdfs://namenode:8040/logs/"。在此路径下直接找到的所有文件都将在发现时进行处理。
• 可以提供 POSIX glob 模式，例如 "hdfs://namenode:8040/logs/2017/*"。在这里，DStream 将包含与该模式匹配的目录中的所有文件。也就是说：它是一个目录模式，而不是目录中文件的模式。
• 所有文件必须采用相同的数据格式。
• 文件所属时间段取决于其修改时间，而不是创建时间。
• 文件经过处理后，在当前窗口内对文件的更改不会导致重新读取该文件。也就是说：*忽略更新*。
• 目录下的文件越多，扫描更改所需的时间就越长，即使没有文件被修改也是如此。
• 如果使用通配符来标识目录，例如 "hdfs://namenode:8040/logs/2016-*"，则将整个目录重命名为与路径匹配会将该目录添加到受监控目录列表中。只有修改时间在当前窗口内的目录中的文件才会包含在流中。
• 调用 FileSystem.setTimes() 来修复时间戳是一种即使文件内容没有更改也能在以后的窗口中选取文件的方法。

使用对象存储作为数据源

像 HDFS 这样的“完整”文件系统倾向于在创建输出流后立即设置文件的修改时间。当文件被打开时，甚至在数据完全写入之前，它可能就被包含在 DStream 中 - 之后，在同一窗口内对文件的更新将被忽略。也就是说：更改可能会被遗漏，数据可能会从流中省略。

为了保证在窗口中获取更改，请将文件写入未监控的目录，然后在输出流关闭后立即将其重命名到目标目录。如果重命名的文件在其创建窗口期间出现在扫描的目标目录中，则将获取新数据。

相比之下，像 Amazon S3 和 Azure 存储这样的对象存储通常具有缓慢的重命名操作，因为数据实际上是被复制的。此外，重命名的对象可能具有 rename() 操作的时间作为其修改时间，因此可能不被视为原始创建时间所暗示的窗口的一部分。

需要针对目标对象存储进行仔细的测试，以验证存储的时间戳行为是否与 Spark Streaming 预期的一致。直接写入目标目录可能是通过所选对象存储流式传输数据的适当策略。

有关此主题的更多详细信息，请参阅 Hadoop 文件系统规范。

基于自定义接收器的流

可以使用通过自定义接收器接收的数据流创建 DStream。有关更多详细信息，请参阅 自定义接收器指南。

RDD 队列作为流

为了使用测试数据测试 Spark Streaming 应用程序，还可以使用 streamingContext.queueStream(queueOfRDDs) 创建基于 RDD 队列的 DStream。推送到队列中的每个 RDD 都将被视为 DStream 中的一批数据，并像流一样进行处理。

有关来自套接字和文件的流的更多详细信息，请参阅 Scala 的 StreamingContext、Java 的 JavaStreamingContext 和 Python 的 StreamingContext 中相关函数的 API 文档。

高级源

Python API 从 Spark 3.5.1 开始，在这些源中，Kafka 和 Kinesis 在 Python API 中可用。

此类源需要与外部非 Spark 库交互，其中一些库具有复杂的依赖关系（例如，Kafka）。因此，为了最大程度地减少与依赖项版本冲突相关的问题，从这些源创建 DStream 的功能已移至单独的库中，这些库可以在必要时显式 链接。

请注意，这些高级源在 Spark shell 中不可用，因此无法在 shell 中测试基于这些高级源的应用程序。如果您真的想在 Spark shell 中使用它们，则必须下载相应的 Maven 工件的 JAR 及其依赖项，并将其添加到类路径中。

以下是一些高级源。

• **Kafka：** Spark Streaming 3.5.1 与 Kafka 代理版本 0.10 或更高版本兼容。有关更多详细信息，请参阅 Kafka 集成指南。

• **Kinesis：** Spark Streaming 3.5.1 与 Kinesis 客户端库 1.2.1 兼容。有关更多详细信息，请参阅 Kinesis 集成指南。

自定义源

Python API Python 尚不支持此功能。

输入 DStream 也可以从自定义数据源创建。您所要做的就是实现一个用户定义的**接收器**（请参阅下一节以了解它是什么），它可以从自定义源接收数据并将其推送到 Spark。有关详细信息，请参阅 自定义接收器指南。

接收器可靠性

根据数据源的*可靠性*，可以有两种类型的数据源。源（如 Kafka）允许确认传输的数据。如果从这些*可靠*源接收数据的系统正确确认了接收到的数据，则可以确保不会因任何类型的故障而丢失数据。这导致了两种接收器

1. *可靠接收器* - 当数据已被接收并存储在 Spark 中并进行复制时，*可靠接收器* 会向可靠源发送确认。
2. *不可靠接收器* - *不可靠接收器**不会*向源发送确认。这可以用于不支持确认的源，甚至可以用于可靠源，当您不想或不需要进入确认的复杂性时。

有关如何编写可靠接收器的详细信息，请参阅 自定义接收器指南。

DStreams 上的转换

与 RDD 类似，转换允许修改来自输入 DStream 的数据。DStream 支持普通 Spark RDD 上可用的许多转换。以下是一些常见的转换。

其中一些转换值得更详细地讨论。

UpdateStateByKey 操作

updateStateByKey 操作允许您在不断更新状态的同时维护任意状态。要使用此功能，您必须执行两个步骤。

1. 定义状态 - 状态可以是任意数据类型。
2. 定义状态更新函数 - 使用函数指定如何使用先前状态和来自输入流的新值来更新状态。

在每个批次中，Spark 将对所有现有键应用状态更新函数，而不管它们在批次中是否有新数据。如果更新函数返回 None，则键值对将被删除。

让我们用一个例子来说明这一点。假设您要维护在文本数据流中看到的每个单词的运行计数。在这里，运行计数是状态，它是一个整数。我们将更新函数定义为

def updateFunction(newValues, runningCount):
    if runningCount is None:
        runningCount = 0
    return sum(newValues, runningCount)  # add the new values with the previous running count to get the new count

runningCounts = pairs.updateStateByKey(updateFunction)

def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val newCount = ...  // add the new values with the previous running count to get the new count
    Some(newCount)
}

val runningCounts = pairs.updateStateByKey[Int](updateFunction _)

Function2<List<Integer>, Optional<Integer>, Optional<Integer>> updateFunction =
  (values, state) -> {
    Integer newSum = ...  // add the new values with the previous running count to get the new count
    return Optional.of(newSum);
  };

JavaPairDStream<String, Integer> runningCounts = pairs.updateStateByKey(updateFunction);

请注意，使用 updateStateByKey 需要配置检查点目录，这在检查点部分中有详细讨论。

转换操作

transform 操作（及其变体，如 transformWith）允许对 DStream 应用任意的 RDD 到 RDD 函数。它可以用来应用 DStream API 中未公开的任何 RDD 操作。例如，将数据流中的每个批次与另一个数据集连接起来的功能在 DStream API 中没有直接公开。但是，您可以轻松地使用 transform 来做到这一点。这带来了非常强大的可能性。例如，可以通过将输入数据流与预先计算的垃圾邮件信息（也可能使用 Spark 生成）连接起来，然后根据它进行过滤，来进行实时数据清理。

spamInfoRDD = sc.pickleFile(...)  # RDD containing spam information

# join data stream with spam information to do data cleaning
cleanedDStream = wordCounts.transform(lambda rdd: rdd.join(spamInfoRDD).filter(...))

val spamInfoRDD = ssc.sparkContext.newAPIHadoopRDD(...) // RDD containing spam information

val cleanedDStream = wordCounts.transform { rdd =>
  rdd.join(spamInfoRDD).filter(...) // join data stream with spam information to do data cleaning
  ...
}

import org.apache.spark.streaming.api.java.*;
// RDD containing spam information
JavaPairRDD<String, Double> spamInfoRDD = jssc.sparkContext().newAPIHadoopRDD(...);

JavaPairDStream<String, Integer> cleanedDStream = wordCounts.transform(rdd -> {
  rdd.join(spamInfoRDD).filter(...); // join data stream with spam information to do data cleaning
  ...
});

请注意，提供的函数会在每个批处理间隔中被调用。这允许您执行时变 RDD 操作，即 RDD 操作、分区数、广播变量等可以在批次之间更改。

窗口操作

Spark Streaming 还提供*窗口计算*，它允许您对滑动数据窗口应用转换。下图说明了此滑动窗口。

如图所示，每次窗口*滑动*到源 DStream 上时，都会组合并操作落入窗口内的源 RDD，以生成窗口化 DStream 的 RDD。在这种特定情况下，该操作应用于最后 3 个时间单位的数据，并按 2 个时间单位滑动。这表明任何窗口操作都需要指定两个参数。

• 窗口长度 - 窗口的持续时间（图中为 3）。
• 滑动间隔 - 执行窗口操作的间隔（图中为 2）。

这两个参数必须是源 DStream 的批处理间隔的倍数（图中为 1）。

让我们用一个例子来说明窗口操作。假设您想通过每 10 秒生成过去 30 秒数据的字数统计来扩展前面的例子。为此，我们必须对过去 30 秒数据的 (word, 1) 对的 pairs DStream 应用 reduceByKey 操作。这是使用操作 reduceByKeyAndWindow 完成的。

# Reduce last 30 seconds of data, every 10 seconds
windowedWordCounts = pairs.reduceByKeyAndWindow(lambda x, y: x + y, lambda x, y: x - y, 30, 10)

// Reduce last 30 seconds of data, every 10 seconds
val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(30), Seconds(10))

// Reduce last 30 seconds of data, every 10 seconds
JavaPairDStream<String, Integer> windowedWordCounts = pairs.reduceByKeyAndWindow((i1, i2) -> i1 + i2, Durations.seconds(30), Durations.seconds(10));

一些常见的窗口操作如下。所有这些操作都采用上述两个参数 - windowLength 和 slideInterval。

连接操作

最后，值得强调的是，在 Spark Streaming 中执行不同类型的连接是多么容易。

流-流连接

流可以非常容易地与其他流连接。

stream1 = ...
stream2 = ...
joinedStream = stream1.join(stream2)

val stream1: DStream[String, String] = ...
val stream2: DStream[String, String] = ...
val joinedStream = stream1.join(stream2)

JavaPairDStream<String, String> stream1 = ...
JavaPairDStream<String, String> stream2 = ...
JavaPairDStream<String, Tuple2<String, String>> joinedStream = stream1.join(stream2);

在这里，在每个批处理间隔中，stream1 生成的 RDD 将与 stream2 生成的 RDD 连接起来。您还可以执行 leftOuterJoin、rightOuterJoin、fullOuterJoin。此外，对流的窗口进行连接通常非常有用。那也很容易。

windowedStream1 = stream1.window(20)
windowedStream2 = stream2.window(60)
joinedStream = windowedStream1.join(windowedStream2)

val windowedStream1 = stream1.window(Seconds(20))
val windowedStream2 = stream2.window(Minutes(1))
val joinedStream = windowedStream1.join(windowedStream2)

JavaPairDStream<String, String> windowedStream1 = stream1.window(Durations.seconds(20));
JavaPairDStream<String, String> windowedStream2 = stream2.window(Durations.minutes(1));
JavaPairDStream<String, Tuple2<String, String>> joinedStream = windowedStream1.join(windowedStream2);

流-数据集连接

这在前面解释 DStream.transform 操作时已经展示过了。下面是将窗口化流与数据集连接的另一个示例。

dataset = ... # some RDD
windowedStream = stream.window(20)
joinedStream = windowedStream.transform(lambda rdd: rdd.join(dataset))

val dataset: RDD[String, String] = ...
val windowedStream = stream.window(Seconds(20))...
val joinedStream = windowedStream.transform { rdd => rdd.join(dataset) }

JavaPairRDD<String, String> dataset = ...
JavaPairDStream<String, String> windowedStream = stream.window(Durations.seconds(20));
JavaPairDStream<String, String> joinedStream = windowedStream.transform(rdd -> rdd.join(dataset));

事实上，您还可以动态更改要连接的数据集。提供给 transform 的函数在每个批处理间隔都会被评估，因此将使用 dataset 引用指向的当前数据集。

DStream 转换的完整列表可在 API 文档中找到。对于 Scala API，请参阅 DStream 和 PairDStreamFunctions。对于 Java API，请参阅 JavaDStream 和 JavaPairDStream。对于 Python API，请参阅 DStream。

DStreams 上的输出操作

输出操作允许将 DStream 的数据推送到外部系统，如数据库或文件系统。由于输出操作实际上允许外部系统使用转换后的数据，因此它们会触发所有 DStream 转换的实际执行（类似于 RDD 的操作）。目前，定义了以下输出操作

使用 foreachRDD 的设计模式

dstream.foreachRDD 是一个强大的原语，允许将数据发送到外部系统。但是，了解如何正确有效地使用此原语非常重要。以下是一些要避免的常见错误。

通常，将数据写入外部系统需要创建一个连接对象（例如，到远程服务器的 TCP 连接）并使用它将数据发送到远程系统。为此，开发人员可能会无意中尝试在 Spark 驱动程序上创建一个连接对象，然后尝试在 Spark 工作器中使用它来保存 RDD 中的记录。例如（在 Scala 中），

def sendRecord(rdd):
    connection = createNewConnection()  # executed at the driver
    rdd.foreach(lambda record: connection.send(record))
    connection.close()

dstream.foreachRDD(sendRecord)

dstream.foreachRDD { rdd =>
  val connection = createNewConnection()  // executed at the driver
  rdd.foreach { record =>
    connection.send(record) // executed at the worker
  }
}

dstream.foreachRDD(rdd -> {
  Connection connection = createNewConnection(); // executed at the driver
  rdd.foreach(record -> {
    connection.send(record); // executed at the worker
  });
});

这是不正确的，因为这需要将连接对象序列化并从驱动程序发送到工作器。此类连接对象很少可以在机器之间传输。此错误可能表现为序列化错误（连接对象不可序列化）、初始化错误（需要在工作器上初始化连接对象）等。正确的解决方案是在工作器上创建连接对象。

但是，这可能会导致另一个常见错误 - 为每条记录创建一个新连接。例如，

def sendRecord(record):
    connection = createNewConnection()
    connection.send(record)
    connection.close()

dstream.foreachRDD(lambda rdd: rdd.foreach(sendRecord))

dstream.foreachRDD { rdd =>
  rdd.foreach { record =>
    val connection = createNewConnection()
    connection.send(record)
    connection.close()
  }
}

dstream.foreachRDD(rdd -> {
  rdd.foreach(record -> {
    Connection connection = createNewConnection();
    connection.send(record);
    connection.close();
  });
});

通常，创建连接对象会占用时间和资源。因此，为每条记录创建和销毁连接对象可能会导致不必要的高开销，并可能显着降低系统的整体吞吐量。更好的解决方案是使用 rdd.foreachPartition - 创建一个连接对象，并使用该连接发送 RDD 分区中的所有记录。

def sendPartition(iter):
    connection = createNewConnection()
    for record in iter:
        connection.send(record)
    connection.close()

dstream.foreachRDD(lambda rdd: rdd.foreachPartition(sendPartition))

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    val connection = createNewConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    connection.close()
  }
}

dstream.foreachRDD(rdd -> {
  rdd.foreachPartition(partitionOfRecords -> {
    Connection connection = createNewConnection();
    while (partitionOfRecords.hasNext()) {
      connection.send(partitionOfRecords.next());
    }
    connection.close();
  });
});

这会将连接创建开销分摊到许多记录上。

最后，可以通过在多个 RDD/批次中重用连接对象来进一步优化这一点。可以维护一个静态连接对象池，而不是在将多个批次的 RDD 推送到外部系统时重复使用，从而进一步减少开销。

def sendPartition(iter):
    # ConnectionPool is a static, lazily initialized pool of connections
    connection = ConnectionPool.getConnection()
    for record in iter:
        connection.send(record)
    # return to the pool for future reuse
    ConnectionPool.returnConnection(connection)

dstream.foreachRDD(lambda rdd: rdd.foreachPartition(sendPartition))

dstream.foreachRDD { rdd =>
  rdd.foreachPartition { partitionOfRecords =>
    // ConnectionPool is a static, lazily initialized pool of connections
    val connection = ConnectionPool.getConnection()
    partitionOfRecords.foreach(record => connection.send(record))
    ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
  }
}

dstream.foreachRDD(rdd -> {
  rdd.foreachPartition(partitionOfRecords -> {
    // ConnectionPool is a static, lazily initialized pool of connections
    Connection connection = ConnectionPool.getConnection();
    while (partitionOfRecords.hasNext()) {
      connection.send(partitionOfRecords.next());
    }
    ConnectionPool.returnConnection(connection); // return to the pool for future reuse
  });
});

请注意，池中的连接应按需延迟创建，并在一段时间未使用后超时。这实现了向外部系统发送数据的最高效率。

其他需要注意的点

• 与 RDD 通过 RDD 操作延迟执行类似，DStream 也通过输出操作延迟执行。具体来说，DStream 输出操作内的 RDD 操作会强制处理接收到的数据。因此，如果您的应用程序没有任何输出操作，或者只有像 dstream.foreachRDD() 这样的输出操作，但其中没有任何 RDD 操作，那么将不会执行任何操作。系统只会接收数据并将其丢弃。

• 默认情况下，输出操作会逐个执行。并且它们按照在应用程序中定义的顺序执行。

DataFrame 和 SQL 操作

您可以轻松地在流数据上使用 DataFrame 和 SQL 操作。您必须使用 StreamingContext 所使用的 SparkContext 创建一个 SparkSession。此外，必须以能够在驱动程序故障时重启的方式完成此操作。这可以通过创建 SparkSession 的延迟实例化单例实例来完成。以下示例对此进行了演示。它修改了之前的 字数统计示例，以使用 DataFrame 和 SQL 生成字数统计。每个 RDD 都将转换为 DataFrame，注册为临时表，然后使用 SQL 进行查询。

# Lazily instantiated global instance of SparkSession
def getSparkSessionInstance(sparkConf):
    if ("sparkSessionSingletonInstance" not in globals()):
        globals()["sparkSessionSingletonInstance"] = SparkSession \
            .builder \
            .config(conf=sparkConf) \
            .getOrCreate()
    return globals()["sparkSessionSingletonInstance"]

...

# DataFrame operations inside your streaming program

words = ... # DStream of strings

def process(time, rdd):
    print("========= %s =========" % str(time))
    try:
        # Get the singleton instance of SparkSession
        spark = getSparkSessionInstance(rdd.context.getConf())

        # Convert RDD[String] to RDD[Row] to DataFrame
        rowRdd = rdd.map(lambda w: Row(word=w))
        wordsDataFrame = spark.createDataFrame(rowRdd)

        # Creates a temporary view using the DataFrame
        wordsDataFrame.createOrReplaceTempView("words")

        # Do word count on table using SQL and print it
        wordCountsDataFrame = spark.sql("select word, count(*) as total from words group by word")
        wordCountsDataFrame.show()
    except:
        pass

words.foreachRDD(process)

/** DataFrame operations inside your streaming program */

val words: DStream[String] = ...

words.foreachRDD { rdd =>

  // Get the singleton instance of SparkSession
  val spark = SparkSession.builder.config(rdd.sparkContext.getConf).getOrCreate()
  import spark.implicits._

  // Convert RDD[String] to DataFrame
  val wordsDataFrame = rdd.toDF("word")

  // Create a temporary view
  wordsDataFrame.createOrReplaceTempView("words")

  // Do word count on DataFrame using SQL and print it
  val wordCountsDataFrame =
    spark.sql("select word, count(*) as total from words group by word")
  wordCountsDataFrame.show()
}

/** Java Bean class for converting RDD to DataFrame */
public class JavaRow implements java.io.Serializable {
  private String word;

  public String getWord() {
    return word;
  }

  public void setWord(String word) {
    this.word = word;
  }
}

...

/** DataFrame operations inside your streaming program */

JavaDStream<String> words = ...

words.foreachRDD((rdd, time) -> {
  // Get the singleton instance of SparkSession
  SparkSession spark = SparkSession.builder().config(rdd.sparkContext().getConf()).getOrCreate();

  // Convert RDD[String] to RDD[case class] to DataFrame
  JavaRDD<JavaRow> rowRDD = rdd.map(word -> {
    JavaRow record = new JavaRow();
    record.setWord(word);
    return record;
  });
  DataFrame wordsDataFrame = spark.createDataFrame(rowRDD, JavaRow.class);

  // Creates a temporary view using the DataFrame
  wordsDataFrame.createOrReplaceTempView("words");

  // Do word count on table using SQL and print it
  DataFrame wordCountsDataFrame =
    spark.sql("select word, count(*) as total from words group by word");
  wordCountsDataFrame.show();
});

您还可以从不同的线程（即与正在运行的 StreamingContext 异步）对流数据上定义的表运行 SQL 查询。只需确保将 StreamingContext 设置为记住足够数量的流数据，以便查询可以运行。否则，StreamingContext（它不知道任何异步 SQL 查询）将在查询完成之前删除旧的流数据。例如，如果您要查询最后一个批次，但您的查询可能需要 5 分钟才能运行，则调用 streamingContext.remember(Minutes(5))（在 Scala 中，或其他语言中的等效项）。

有关 DataFrame 的更多信息，请参阅 DataFrame 和 SQL 指南。

MLlib 操作

您还可以轻松使用 MLlib 提供的机器学习算法。首先，有一些流式机器学习算法（例如 流式线性回归、流式 KMeans 等）可以同时从流数据中学习并在流数据上应用模型。除此之外，对于更大类的机器学习算法，您可以离线学习学习模型（即使用历史数据），然后在线将模型应用于流数据。有关更多详细信息，请参阅 MLlib 指南。

缓存/持久化

与 RDD 类似，DStream 也允许开发人员将流的数据持久化到内存中。也就是说，在 DStream 上使用 persist() 方法将自动将该 DStream 的每个 RDD 持久化到内存中。如果 DStream 中的数据将被计算多次（例如，对相同数据进行多次操作），这将非常有用。对于像 reduceByWindow 和 reduceByKeyAndWindow 这样的基于窗口的操作以及像 updateStateByKey 这样的基于状态的操作，这都是隐式成立的。因此，由基于窗口的操作生成的 DStream 会自动持久化到内存中，而无需开发人员调用 persist()。

对于通过网络接收数据的输入流（例如 Kafka、套接字等），默认的持久化级别设置为将数据复制到两个节点以实现容错。

请注意，与 RDD 不同，DStream 的默认持久化级别会将数据序列化保存在内存中。这将在 性能调优 部分进一步讨论。有关不同持久化级别的更多信息，请参阅 Spark 编程指南。

检查点

流应用程序必须全天候运行，因此必须能够抵御与应用程序逻辑无关的故障（例如系统故障、JVM 崩溃等）。为此，Spark Streaming 需要将足够的信息_检查点_到容错存储系统，以便能够从故障中恢复。有两种类型的数据需要进行检查点。

• _元数据检查点_ - 将定义流计算的信息保存到容错存储（如 HDFS）中。这用于从运行流应用程序驱动程序的节点故障中恢复（稍后详细讨论）。元数据包括
  • _配置_ - 用于创建流应用程序的配置。
  • _DStream 操作_ - 定义流应用程序的 DStream 操作集。
  • _未完成的批次_ - 其作业已排队但尚未完成的批次。
• _数据检查点_ - 将生成的 RDD 保存到可靠的存储中。这在某些跨多个批次组合数据的_有状态_转换中是必需的。在此类转换中，生成的 RDD 依赖于先前批次的 RDD，这会导致依赖链的长度随着时间的推移而不断增加。为了避免恢复时间（与依赖链成正比）出现这种无限增加的情况，会定期将有状态转换的中间 RDD _检查点_到可靠的存储（例如 HDFS）中，以切断依赖链。

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

何时启用检查点

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

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

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

如何配置检查点

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

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

# 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()

// 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()

// 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();

除了使用 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.12及其所有传递依赖项。

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

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

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

• 配置预写日志 - 从 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.receiver.maxRate，有关直接 Kafka 方法，请参阅spark.streaming.kafka.maxRatePerPartition。在 Spark 1.5 中，我们引入了一项称为_背压_的功能，该功能无需设置此速率限制，因为 Spark Streaming 会自动确定速率限制并在处理条件发生变化时动态调整它们。可以通过将配置参数spark.streaming.backpressure.enabled设置为true来启用此背压。

升级应用程序代码

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

• 升级后的 Spark Streaming 应用程序与现有应用程序并行启动和运行。一旦新的应用程序（接收与旧应用程序相同的数据）预热完毕并准备好投入使用，就可以关闭旧的应用程序。请注意，这可以针对支持将数据发送到两个目标（即，较早的应用程序和升级后的应用程序）的数据源完成。

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

监控应用程序

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

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

• 处理时间 - 处理每批数据的时间。
• 调度延迟 - 批次在队列中等待处理先前批次完成的时间。

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

还可以使用StreamingListener接口监控 Spark Streaming 程序的进度，该接口允许您获取接收器状态和处理时间。请注意，这是一个开发者 API，将来可能会对其进行改进（即，报告更多信息）。

性能调优

要在集群上获得 Spark Streaming 应用程序的最佳性能需要进行一些调整。本节介绍了许多可以调整的参数和配置，以提高应用程序的性能。在较高的层次上，您需要考虑两件事

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

2. 设置正确的批大小，以便可以以接收数据的速度处理数据批次（也就是说，数据处理与数据提取保持同步）。

减少批处理时间

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

数据接收的并行度

通过网络（如 Kafka、套接字等）接收数据需要对数据进行反序列化并存储在 Spark 中。如果数据接收成为系统的瓶颈，请考虑并行化数据接收。请注意，每个输入 DStream 都会创建一个接收器（在工作机器上运行），该接收器接收单个数据流。因此，可以通过创建多个输入 DStream 并将它们配置为从源接收数据流的不同分区来实现接收多个数据流。例如，可以将接收两个数据主题的单个 Kafka 输入 DStream 拆分为两个 Kafka 输入流，每个流仅接收一个主题。这将运行两个接收器，允许并行接收数据，从而提高整体吞吐量。可以将这些多个 DStream 合并在一起以创建单个 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();

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

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

有关直接流，请参阅 Spark Streaming + Kafka 集成指南

数据处理中的并行度

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

数据序列化

可以通过调整序列化格式来减少数据序列化的开销。在流式传输的情况下，有两种类型的数据正在被序列化。

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

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

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

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

任务启动开销

如果每秒启动的任务数量很多（例如，每秒 50 个或更多），那么将任务发送到执行器的开销可能会很大，并且将难以实现亚秒级延迟。可以通过以下更改来减少开销

• 执行模式：与细粒度 Mesos 模式相比，在独立模式或粗粒度 Mesos 模式下运行 Spark 会导致更好的任务启动时间。有关更多详细信息，请参阅 在 Mesos 上运行指南。

这些更改可以将批处理时间减少 100 毫秒，从而使亚秒级批处理大小成为可能。

设置正确的批处理间隔

为了使在集群上运行的 Spark Streaming 应用程序保持稳定，系统应该能够以接收数据的速度处理数据。换句话说，数据批次的处理速度应该与其生成速度一样快。可以通过 监控 流式 Web UI 中的处理时间来确定应用程序是否如此，其中批处理时间应小于批次间隔。

根据流式计算的性质，所使用的批次间隔可能会对应用程序在固定集群资源集上可以维持的数据速率产生重大影响。例如，让我们考虑前面的 WordCountNetwork 示例。对于特定的数据速率，系统可能能够每 2 秒报告一次字数统计（即 2 秒的批次间隔），但不能每 500 毫秒报告一次。因此，需要设置批次间隔，以便可以维持生产中的预期数据速率。

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

内存调优

在 调整指南 中详细讨论了 Spark 应用程序的内存使用和 GC 行为的调整。强烈建议您阅读该内容。在本节中，我们将专门针对 Spark Streaming 应用程序讨论一些调整参数。

Spark Streaming 应用程序所需的集群内存量很大程度上取决于所使用的转换类型。例如，如果您想对最后 10 分钟的数据使用窗口操作，那么您的集群应该有足够的内存来在内存中保存 10 分钟的数据。或者，如果您想将 updateStateByKey 与大量键一起使用，那么所需的内存将会很高。相反，如果您想执行简单的 map-filter-store 操作，那么所需的内存将会很低。

通常，由于通过接收器接收的数据使用 StorageLevel.MEMORY_AND_DISK_SER_2 存储，因此不适合内存的数据将溢出到磁盘。这可能会降低流式应用程序的性能，因此建议根据流式应用程序的需要提供足够的内存。最好尝试查看小规模的内存使用情况并据此进行估计。

内存调整的另一个方面是垃圾收集。对于需要低延迟的流式应用程序来说，不希望出现由 JVM 垃圾收集引起的长暂停。

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

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

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

• CMS 垃圾收集器：强烈建议使用并发标记清除 GC 来保持与 GC 相关的暂停始终较低。尽管已知并发 GC 会降低系统的整体处理吞吐量，但仍然建议使用它来实现更一致的批处理时间。确保在驱动程序（在 spark-submit 中使用 --driver-java-options）和执行器（使用 Spark 配置 spark.executor.extraJavaOptions）上都设置了 CMS GC。

• 其他提示：要进一步减少 GC 开销，请尝试以下更多提示。

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

需要记住的要点

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

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

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

• 块上的映射任务在具有块的执行器（接收块的执行器和复制块的执行器）中处理，而不管块间隔如何，除非启动了非本地调度。更大的 blockinterval 意味着更大的块。spark.locality.wait 的高值会增加在本地节点上处理块的可能性。需要在这两个参数之间找到平衡，以确保更大的块在本地处理。

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

• 如果您有两个 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 转换都是确定性的，则最终转换后的 RDD 中的数据将始终相同，而与 Spark 集群中的故障无关。

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

1. 已接收并复制的数据 - 此数据在单个工作节点发生故障后仍然存在，因为它的副本存在于其他某个节点上。
2. 已接收但已缓冲以进行复制的数据 - 由于这没有被复制，因此恢复此数据的唯一方法是从源重新获取它。

此外，我们应该关注两种故障

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

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

定义

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

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

基本语义

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

1. 接收数据：使用接收器或其他方式从源接收数据。

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

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

如果流式应用程序要实现端到端的精确一次保证，则每个步骤都必须提供精确一次保证。也就是说，每条记录必须被精确接收一次，精确转换一次，并精确推送到下游系统一次。让我们在 Spark Streaming 的上下文中了解这些步骤的语义。

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

2. 转换数据：由于 RDD 提供的保证，所有已接收的数据都将被*精确处理一次*。即使出现故障，只要接收到的输入数据可访问，最终转换后的 RDD 将始终具有相同的内容。

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

接收数据的语义

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

使用文件

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

使用基于接收器的源

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

1. 可靠接收器 - 这些接收器仅在确保已复制接收到的数据后才确认可靠源。如果此类接收器发生故障，则源将不会收到已缓冲（未复制）数据的确认。因此，如果接收器重新启动，源将重新发送数据，并且不会因故障而丢失数据。
2. 不可靠接收器 - 此类接收器*不*发送确认，因此在因工作程序或驱动程序故障而发生故障时*可能*会丢失数据。

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

为了避免这种过去接收数据的丢失，Spark 1.2 引入了*预写日志*，它将接收到的数据保存到容错存储中。在启用了预写日志和可靠接收器的情况下，数据丢失为零。在语义方面，它提供了至少一次的保证。

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

使用 Kafka Direct API

在 Spark 1.3 中，我们引入了一个新的 Kafka Direct API，它可以确保 Spark Streaming 恰好一次接收所有 Kafka 数据。除此之外，如果您实现恰好一次输出操作，则可以实现端到端的恰好一次保证。这种方法在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 文档
  • Scala 文档
    • StreamingContext 和 DStream
    • KafkaUtils、KinesisUtils，
  • Java 文档
    • JavaStreamingContext、JavaDStream 和 JavaPairDStream
    • KafkaUtils、KinesisUtils
  • Python 文档
    • StreamingContext 和 DStream
• 更多示例请参见 Scala、Java 和 Python
• 介绍 Spark Streaming 的论文和视频。