Spark Streaming Programming Guide

我就是我 2023-07-11 15:13 37阅读 0赞

**Table of Contents**

简述

简易样例

基本概念

依赖

StreamingContext初始化

Discretized Streams (DStreams)

输入数据流和接收器

基础Sources

高级Sources

自定义Sources

Receiver 可靠性

DStreams上的算子

UpdateStateByKey Operation

Transform Operation

Window Operations

Join Operations

DStreams上的输出操作

使用foreachRDD的设计模式

DataFrame和SQL操作

MLlib Operations

Caching / Persistence

Checkpointing

When to enable Checkpointing

How to configure Checkpointing

Accumulators, Broadcast Variables, and Checkpoints

Deploying Applications

Requirements

Upgrading Application Code

Monitoring Applications

Performance Tuning

Reducing the Batch Processing Times

Level of Parallelism in Data Receiving

Level of Parallelism in Data Processing

Data Serialization

Task Launching Overheads

Setting the Right Batch Interval

Memory Tuning

Fault-tolerance Semantics

Background

容错Definitions

Basic Semantics

Semantics of Received Data

With Files

With Receiver-based Sources

With Kafka Direct API

Semantics of output operations

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

# 简述 #

Spark Streaming 是核心 Spark API 的扩展，支持可伸缩、高吞吐量、容错的实时数据流处理。数据可以从许多来源获取，如Kafka、Flume、Kinesis 或 TCP sockets，可以使用复杂的算法处理数据，这些算法用高级函数表示，如 map、reduce、join和window。最后，处理后的数据可以推送到文件系统、数据库和活动仪表板。实际上，您可以将 Spark 的机器学习和图形处理算法应用于数据流。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70][]

在内部，它是这样工作的。Spark Streaming 接受实时输入数据流，并将数据分成批次，然后由 Spark engine 处理，以批量生成最终的结果流。

![20200306115055525.png][]

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

本指南向您展示了如何使用 DStreams 编写 Spark 流程序。您可以用 Scala、Java或 Python(在Spark 1.2中引入)编写 Spark 流程序，所有这些都在本指南中介绍。

> 注意:有一些api是不同的，或者在Python中不可用的。

# 简易样例 #

在详细介绍如何编写自己的 Spark Streaming 程序之前，让我们先快速了解一下简单的 Spark 流程序是什么样子的。假设我们想要计算从监听 TCP 套接字的数据服务器接收到的文本数据中的单词数。你所需要做的就是如下所示。

首先，我们将 Spark 流类的名称和一些从 StreamingContext 的隐式转换导入到我们的环境中，以便向我们需要的其他类(如DStream)添加有用的方法。StreamingContext 是所有流功能的主要入口点。我们使用两个执行线程创建一个本地StreamingContext，批处理间隔为1秒。

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 from a starvation scenario.
    
    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    val ssc = new StreamingContext(conf, Seconds(1))

使用这个上下文，我们可以创建一个表示来自 TCP 源的流数据的 DStream，指定为主机名(例如localhost)和端口(例如9999)。

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

这行 DStream 表示将从数据服务器接收的数据流。DStream 中的每个记录都是一行文本。接下来，我们希望按空格字符将行分割为单词。

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

flatMap 是一个一对多的 DStream 操作，它通过从源 DStream 中的每个记录生成多个新记录来创建一个新的 DStream。在本例中，每一行将被分成多个单词，单词流表示为单词 DStream。接下来，我们要数这些单词。

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

单词DStream被进一步映射(一对一转换)到一个(单词，1)对的 DStream，然后对其进行简化以获得每批数据中单词的频率。最后，wordcount .print()将打印每秒生成的一些计数。

> 请注意，当这些行被执行时，Spark Streaming 仅设置它将在启动时执行的计算，而没有真正的处理已经启动。为了在设置完所有转换之后开始处理，我们最后调用
> 
>     ssc.start()             // Start the computation
>     ssc.awaitTermination()  // Wait for the computation to terminate

如果您已经下载并构建了 Spark，您可以按如下方式运行这个示例。您首先需要使用 Netcat(在大多数类unix系统中可以找到的一个小实用程序)作为数据服务器运行

$ nc -lk 9999

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

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70 1][]

# 基本概念 #

接下来，我们将继续简单的示例，详细介绍 Spark 流的基础知识。

## 依赖 ##

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

<dependency>
 <groupId>org.apache.spark</groupId>
 <artifactId>spark-streaming_2.11</artifactId>
 <version>2.1.1</version>
 </dependency>

要从 Kafka、Flume 和 Kinesis 等 Spark 流核心 API 中没有的数据源中获取数据，您必须将相应的工件 Spark -stream -xyz\_2.11添加到依赖项中。例如，一些常见的例子如下。

<table> 
 <tbody> 
 <tr> 
 <th>Source</th> 
 <th>Artifact</th> 
 </tr> 
 <tr> 
 <td>Kafka</td> 
 <td>spark-streaming-kafka-0-8_2.11</td> 
 </tr> 
 <tr> 
 <td>Flume</td> 
 <td>spark-streaming-flume_2.11</td> 
 </tr> 
 <tr> 
 <td>Kinesis</td> 
 <td>spark-streaming-kinesis-asl_2.11 [Amazon Software License]</td> 
 </tr> 
 </tbody> 
</table>

## StreamingContext初始化 ##

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

可以从 SparkConf 对象创建 StreamingContext 对象。

import org.apache.spark._
    import org.apache.spark.streaming._
    
    val conf = new SparkConf().setAppName(appName).setMaster(master)
    val ssc = new StreamingContext(conf, Seconds(1))

appName参数是应用程序在集群UI上显示的名称。master是一个Spark、Mesos或YARN集群URL，或者一个特殊的“local\[\*\]”字符串，在本地模式下运行。

实际上，在集群上运行时，您不希望在程序中写死 master，而是使用 spark-submit 启动应用程序并在那里接收它。但是，对于本地测试和单元测试，您可以通过“local\[\*\]”来运行 Spark Streaming in-process(检测本地系统中的内核数量)。

批处理间隔必须根据应用程序的延迟需求和可用的集群资源来设置。

StreamingContext 对象也可以从现有的 SparkContext 对象创建。

import org.apache.spark.streaming._
    
    val sc = ...                // existing SparkContext
    val ssc = new StreamingContext(sc, Seconds(1))

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

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

**注意：**

1.  一旦上下文启动，就不能设置或添加新的流计算。
2.  一旦上下文被停止，就不能重新启动它。
3.  JVM 中只能同时激活一个 StreamingContext。
4.  StreamingContext 上的 stop()也会停止 SparkContext。要仅停止 StreamingContext，请将名为 stopSparkContext 的 stop()的可选参数设置为 false。
5.  只要在创建下一个 StreamingContext 之前停止前一个 StreamingContext(不停止SparkContext)，就可以重用 SparkContext来创建多个 StreamingContext。

## Discretized Streams (DStreams) ##

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

![202003061758450.png][]

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70 2][]

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

## 输入数据流和接收器 ##

输入数据流是表示从流媒体源接收的输入数据流的数据流。在这个简单的示例中，行是一个输入DStream，因为它表示从netcat服务器接收到的数据流。每个输入 DStream(本节后面讨论的文件流除外)都与接收方(Scala doc、Java doc)对象相关联，接收方接收来自源的数据并将其存储在Spark内存中进行处理。

Spark流媒体提供了两类内置的流媒体源。

*  Basic Sources：StreamingContext API中直接可用的资源。示例:file systems, and socket connections。
 *  Advanced Sources：像Kafka, Flume, Kinesis等资源可以通过额外的工具类获得。如链接部分所述，这些需要针对额外依赖项进行链接。

注意，如果希望在流应用程序中并行接收多个数据流，可以创建多个输入DStreams(将在性能调优一节中进一步讨论)。这将创建多个接收器，同时接收多个数据流。但是请注意，Spark worker/executor 是一个长时间运行的任务，因此它占用分配给 Spark  流应用程序的一个核心。因此，重要的是要记住，Spark流应用程序需要分配足够的 core(或线程，如果在本地运行)来处理接收到的数据。

**牢记：**

*  在本地运行 Spark 流程序时，不要使用“local”或“local\[1\]”作为主URL。这两种方法都意味着只有一个线程将用于在本地运行任务。如果您正在使用基于接收器的输入DStream(例如sockets、Kafka、Flume等)，那么将使用单个线程来运行接收器，不留下任何线程来处理接收到的数据。因此，在本地运行时，始终使用“local\[n\]”作为主URL，其中要运行n个接收方(有关如何设置主接收方的信息，请参阅[https://blog.csdn.net/zpf\_940810653842/article/details/103372744\#spark-properties][https_blog.csdn.net_zpf_940810653842_article_details_103372744_spark-properties])。
 *  将逻辑扩展到在集群上运行，分配给Spark流应用程序的内核数量必须大于接收器的数量。否则，系统将接收数据，但无法处理它。

### 基础Sources ###

我们已经在这个简易的示例中看到了ssc.socketTextStream(…)，它根据通过 TCP 套接字连接接收的文本数据创建 DStream。除了套接字之外，StreamingContext API 还提供了将文件创建为输入源的方法。

**文件流：**为了从任何与HDFS API(即HDFS、S3、NFS等)兼容的文件系统上的文件中读取数据，可以创建一个DStream:

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

Spark 流将监视目录 dataDirectory 并处理在该目录中创建的任何文件(不支持在嵌套目录中编写的文件)。请注意,

*  这些文件必须具有相同的数据格式。
 *  这些文件必须在 dataDirectory 中创建，方法是自动地将它们移动或重命名到数据目录中。
 *  文件一旦被移动，就不能被改变。因此，如果文件是连续追加的，则不会读取新数据。

对于简单的文本文件，有一个更简单的方法 streamingContext.textFileStream(dataDirectory)。文件流不需要运行接收器，因此不需要分配 core。

> Python API中没有fileStream，只有textFileStream可用。

**基于自定义接收器的流：**DStreams 可以使用通过自定义接收器接收的数据流创建。有关详细信息，请参阅

[Spark Streaming Custom Receivers][]。

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

### 高级Sources ###

在这些源代码中，Kafka、Kinesis 和 Flume都可以在 Python API中找到。

这类源需要与外部非 spark 库交互，其中一些具有复杂的依赖关系(例如Kafka和Flume)。因此，为了最小化与版本依赖冲突相关的问题，从这些源创建 DStreams 的功能已经转移到单独的库中，在必要时可以显式地链接到这些库。

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

Kafka：[http://spark.apache.org/docs/2.1.1/streaming-kafka-integration.html][http_spark.apache.org_docs_2.1.1_streaming-kafka-integration.html]

......

### 自定义Sources ###

> 这在Python中还不支持。

还可以从自定义数据源创建输入DStreams。您所要做的就是实现一个用户定义的接收器(请参阅下一节了解它是什么)，它可以接收来自自定义源的数据并将其推入Spark。有关详细信息，请参阅 [Spark Streaming Custom Receivers][]。

### Receiver 可靠性 ###

基于数据源的可靠性，可以有两种数据源。数据源(如Kafka和Flume)允许确认传输的数据。如果从这些可靠来源接收数据的系统正确地确认接收到的数据，则可以确保不会由于任何类型的故障而丢失任何数据。这就导致了两种类型的 receivers：

1.  Reliable Receiver - 当数据已被接收并存储在带有复制的Spark中时，可靠的接收方正确地向可靠的源发送确认。
2.  Unreliable Reciver - 不可靠的接收器不向源发送确认。这可以用于不支持确认的源，甚至可以用于不希望或不需要进入确认复杂性的可靠源。

## DStreams上的算子 ##

与 RDDs类似，转换允许修改输入 DStream 中的数据。DStreams 支持许多在普通 Spark RDD 上可用的转换。一些常见的例子如下。

<table> 
 <tbody> 
 <tr> 
 <th>Transformation</th> 
 <th>Meaning</th> 
 </tr> 
 <tr> 
 <td>map(func)</td> 
 <td>通过函数func传递源 DStream 的每个元素来返回一个新的 DStream。</td> 
 </tr> 
 <tr> 
 <td>flatMap(func)</td> 
 <td>与map类似，但是每个输入项可以映射到0或多个输出项。</td> 
 </tr> 
 <tr> 
 <td>filter(func)</td> 
 <td>通过只选择 func 返回true的源 DStream 的记录来返回一个新的DStream。</td> 
 </tr> 
 <tr> 
 <td>repartition(numPartitions)</td> 
 <td>通过创建更多或更少的分区来改变 DStream 中的并行度。</td> 
 </tr> 
 <tr> 
 <td>union(otherStream)</td> 
 <td>返回一个新的 DStream，它包含源 DStream 和 otherDStream 中元素的并集。</td> 
 </tr> 
 <tr> 
 <td>count()</td> 
 <td>通过计算源 DStream 的每个RDD中的元素数量，返回一个新的单元素 RDDs DStream。</td> 
 </tr> 
 <tr> 
 <td>reduce(func)</td> 
 <td>通过使用函数func(它接受两个参数并返回一个参数)聚合源 DStream 的每个 RDD中的元素，返回一个新的单元素 RDDs DStream。这个函数应该是结合律和交换律，这样才能并行计算。</td> 
 </tr> 
 <tr> 
 <td>countByValue()</td> 
 <td>当对类型为K的元素的DStream调用时，返回一个新的DStream (K, Long)对，其中每个键的值是它在源DStream的每个RDD中的频率。</td> 
 </tr> 
 <tr> 
 <td>reduceByKey(func, [numTasks])</td> 
 <td>当在(K, V)对的DStream上调用时，返回一个新的(K, V)对的DStream，其中每个键的值使用给定的reduce函数进行聚合。注意:在默认情况下，这将使用Spark的默认并行任务数(本地模式为2，而在集群模式下，该数量由配置属性Spark .default.parallelism决定)来进行分组。您可以传递一个可选的numTasks参数来设置不同数量的任务。</td> 
 </tr> 
 <tr> 
 <td>join(otherStream, [numTasks])</td> 
 <td>当调用两个DStream (K, V)和(K, W)对时，返回一个新的DStream (K， (V, W))对，每个键的所有对的元素。</td> 
 </tr> 
 <tr> 
 <td>cogroup(otherStream, [numTasks])</td> 
 <td>当调用一个(K, V)和(K, W)对的DStream时，返回一个(K, Seq[V]， Seq[W])元组的新DStream。</td> 
 </tr> 
 <tr> 
 <td>transform(func)</td> 
 <td>通过对源 DStream 的每个RDD应用一个RDD-to-RDD函数来返回一个新的DStream。这可以用来在DStream上执行任意的RDD操作。</td> 
 </tr> 
 <tr> 
 <td>updateStateByKey(func)</td> 
 <td>返回一个新的“状态”DStream，其中通过对键的前一个状态和键的新值应用给定的函数来更新每个键的状态。这可以用来维护每个键的任意状态数据。</td> 
 </tr> 
 </tbody> 
</table>

### UpdateStateByKey Operation ###

updateStateByKey 操作允许您维护任意状态，同时不断地用新信息更新它。要使用它，您必须执行两个步骤。

*  定义状态——状态可以是任意的数据类型。
 *  定义状态更新函数——使用一个函数来指定如何使用输入流中的前一个状态和新值来更新状态。

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

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

def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
        val newCount = ...  // 使用前一个运行的计数添加新值以获得新计数
        Some(newCount)
    }

这将应用于包含单词的 DStream(例如，前面示例中包含(word，1)对的 DStream)。

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

每个单词都将调用 update 函数，newValues 的序列为1(来自(word, 1)对)，runningCount 的序列为前一个计数。

> 注意，使用updateStateByKey需要配置检查点目录

### Transform Operation ###

转换操作(及其变体，如transformWith)允许在DStream上应用任意的RDD-to-RDD函数。它可以用于应用DStream API中没有公开的任何RDD操作。例如，将数据流中的每个批处理与另一个数据集连接的功能并没有直接在DStream API中公开。但是，您可以很容易地使用transform来实现这一点。这带来了非常强大的可能性。例如，可以通过将输入数据流与预先计算的垃圾信息(也可以使用Spark生成)a连接起来，从而进行实时数据清理。

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
      ...
    }

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

### Window Operations ###

Spark 流还提供了窗口计算，它允许您在数据的滑动窗口上应用转换。下图演示了这个滑动窗口。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70 3][]

如图所示，每当窗口在源 DStream 上滑动时，位于窗口内的源 RDDs 就会被合并并操作，以生成加了窗口的 DStream的RDDs。在本例中，操作应用于数据的最后3个时间单位，幻灯片应用于2个时间单位。这表明任何窗口操作都需要指定两个参数。

*  窗口大小：窗口的持续时间(图中为3)。
 *  滑动间隔：窗口操作执行的时间间隔(图中为2)。

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

让我们用一个例子来说明窗口操作。例如，您希望通过每10秒在最后30秒的数据中生成字数计数来扩展前面的示例。为此，我们必须在最后30秒的数据中对(word, 1)对的 DStream 应用 reduceByKey 操作。这是使用 reduceByKeyAndWindow 操作完成的。

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

下面是一些常见的窗口操作。所有这些操作都采用上述两个参数——窗口长度和滑动间隔。

<table> 
 <tbody> 
 <tr> 
 <th>Transformation</th> 
 <th>Meaning</th> 
 </tr> 
 <tr> 
 <td>window(windowLength, slideInterval)</td> 
 <td>返回一个新的DStream，它是基于源DStream的加窗批量计算的。</td> 
 </tr> 
 <tr> 
 <td>countByWindow(windowLength, slideInterval)</td> 
 <td>返回流中元素的滑动窗口计数。</td> 
 </tr> 
 <tr> 
 <td>reduceByWindow(func, windowLength, slideInterval)</td> 
 <td>返回一个新的单元素流，它是通过使用func将流中的元素在一个滑动区间内聚合而创建的。这个函数应该是结合律和交换律，这样才能正确地并行计算。</td> 
 </tr> 
 <tr> 
 <td>reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])</td> 
 <td>当在一个(K, V)对的DStream上调用时，返回一个新的(K, V)对的DStream，其中每个键的值在滑动窗口中使用给定的reduce函数func over batch进行聚合。Note: B默认情况下，这将使用Spark的默认并行任务数量(本地模式为2个，在集群模式下，该数量由配置属性Spark .default.parallelism决定)来进行分组。您可以传递一个可选的numTasks参数来设置不同数量的任务。</td> 
 </tr> 
 <tr> 
 <td>reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])</td> 
 <td> 上面的reduceByKeyAndWindow()的一个更有效的版本，其中每个窗口的reduce值是使用前一个窗口的reduce值递增计算的。这是通过减少进入滑动窗口的新数据和“反向减少”离开窗口的旧数据来实现的。例如，在窗口滑动时“添加”和“减去”键数。 但是，它只适用于“可逆约简函数”，即具有相应的“逆约简”函数的约简函数(取参数invFunc)。与reduceByKeyAndWindow一样，reduce任务的数量可以通过一个可选参数进行配置。 注意，必须启用检查点才能使用此操作。 </td> 
 </tr> 
 <tr> 
 <td>countByValueAndWindow(windowLength, slideInterval, [numTasks])</td> 
 <td>当调用一个(K, V)对的DStream时，返回一个新的(K, Long)对的DStream，其中每个键的值是它在滑动窗口中的频率。与reduceByKeyAndWindow一样，reduce任务的数量可以通过一个可选参数进行配置。</td> 
 </tr> 
 </tbody> 
</table>

### Join Operations ###

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

**Stream-stream joins**

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

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

在这里，在每个批处理间隔中，由stream1生成的RDD将与由stream2生成的RDD相连接。你也可以使用 leftOuterJoin, rightOuterJoin, fullOuterJoin。此外，在流的窗口上进行连接通常非常有用。这也很简单。

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

**Stream-dataset joins**

在前面的explain DStream中已经显示了这一点。变换操作。下面是另一个将窗口化的流与数据集连接起来的示例。

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

实际上，您还可以动态地更改要加入的数据集。提供的转换函数在每个批处理间隔进行评估，因此将使用数据集引用点所指向的当前数据集。

DStream转换的完整列表可以在API文档中找到。有关Scala API，请参见 [DStream][] 和 [PairDStreamFunctions][]。有关Java API，请参见 [JavaDStream][] 和 [JavaPairDStream][]。有关Python API，请参阅 [DStream][DStream 1]。

## DStreams上的输出操作 ##

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

<table> 
 <tbody> 
 <tr> 
 <th>Output Operation</th> 
 <th>Meaning</th> 
 </tr> 
 <tr> 
 <td>print()</td> 
 <td>在运行流应用程序的驱动节点上打印 DStream 中每批数据的前十个元素。这对于开发和调试非常有用。 这在Python API中称为pprint()。</td> 
 </tr> 
 <tr> 
 <td>saveAsTextFiles(prefix, [suffix])</td> 
 <td>将 DStream 的内容保存为文本文件。每个批处理间隔的文件名是根据前缀和后缀“prefix-TIME_IN_MS[.suffix]”生成的。</td> 
 </tr> 
 <tr> 
 <td>saveAsObjectFiles(prefix, [suffix])</td> 
 <td>将这个DStream的内容保存为序列化的Java对象的序列文件。每个批处理间隔的文件名是根据前缀和后缀“prefix-TIME_IN_MS[.suffix]”生成的。 这在Python API中不可用。</td> 
 </tr> 
 <tr> 
 <td>saveAsHadoopFiles(prefix, [suffix])</td> 
 <td>将DStream的内容保存为Hadoop文件。每个批处理间隔的文件名是根据前缀和后缀“prefix-TIME_IN_MS[.suffix]”生成的。 这在Python API中不可用。</td> 
 </tr> 
 <tr> 
 <td>foreachRDD(func)</td> 
 <td>将函数func应用于从流生成的每个RDD的最通用的输出操作符。该函数应该将每个RDD中的数据推送到外部系统，例如将RDD保存到文件中，或者通过网络将其写入数据库。请注意，func函数是在运行流应用程序的驱动程序进程中执行的，并且通常会有RDD动作，这将强制流RDDs的计算。</td> 
 </tr> 
 </tbody> 
</table>

### 使用foreachRDD的设计模式 ###

dstream.foreachRDD 是一个功能强大的原语，它允许将数据发送到外部系统。然而，理解如何正确和有效地使用这个原语是很重要的。要避免的一些常见错误如下。

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

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

这是不正确的，因为这需要将连接对象序列化并从驱动程序发送到工作程序。这样的连接对象很少能跨机器转移。此错误可能表现为序列化错误(连接对象不可序列化)、初始化错误(连接对象需要在工作人员处初始化)等。正确的解决方案是在worker上创建连接对象。

然而，这可能会导致另一个常见错误——为每个记录创建一个新连接。例如,

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

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

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

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

最后，可以通过跨多个RDDs/batch重用连接对象进一步优化这一点。可以维护一个静态的连接对象池，在将多个批的RDDs推送到外部系统时可以重用这些对象，从而进一步减少开销。

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
      }
    }

请注意，池中的连接应该按需延迟创建，如果不使用一段时间就会超时。这实现了向外部系统发送数据的最有效的方式。

**其他需要注意的点：**

*  DStreams 由输出操作延迟执行，就像RDD操作延迟执行rds一样。具体来说，DStream 输出操作中的RDD操作强制处理接收到的数据。因此，如果您的应用程序没有任何输出操作，或者有像 dstream.foreachRDD()这样的输出操作，但是其中没有任何RDD操作，那么什么也不会执行。系统将简单地接收数据并丢弃它。
 *  默认情况下，一次执行一个输出操作。它们是按照在应用程序中定义的顺序执行的。

## DataFrame和SQL操作 ##

您可以轻松地在流数据上使用 DataFrames 和 SQL 操作。您必须使用 StreamingContext 正在使用的 SparkContext 创建一个SparkSession。此外，这样做可以在驱动程序失败时重新启动。这是通过创建一个延迟实例化的 SparkSession 单例实例来实现的。如下面的例子所示。它修改了前面的字数统计示例，以使用 DataFrames 和 SQL 生成字数统计。每个 RDD 都被转换为一个DataFrame，注册为一个临时表，然后使用 SQL 进行查询。

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

您还可以对来自不同线程(即与运行中的 StreamingContext 异步)的流数据上定义的表运行 SQL 查询。只要确保您设置了StreamingContext 来记住足够数量的流数据，以便查询可以运行。否则，StreamingContext (它不知道任何异步SQL查询)将在查询完成之前删除旧的流数据。例如，如果您想要查询最后一批数据，但是您的查询可能需要5分钟才能运行，那么可以调用streamingContext.remember(minutes(5))(在Scala中是这样，或者在其他地方是相同的)

参见 [Spark SQL, DataFrames and Datasets Guide][Spark SQL_ DataFrames and Datasets Guide]，以了解更多关于DataFrames的信息。

## MLlib Operations ##

您还可以轻松地使用由 MLlib 提供的机器学习算法。首先，有流式机器学习算法(如流式线性回归、流式KMeans等)可以同时从流式数据中学习，也可以将模型应用到流式数据中。除此之外，对于更大类别的机器学习算法，您可以离线学习一个学习模型(即使用历史数据)，然后在线将该模型应用于流数据。

## Caching / Persistence ##

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

对于通过网络接收数据的输入流(例如，Kafka、Flume、sockets等)，默认的持久性级别被设置为将数据复制到两个节点以实现容错。

注意，与RDDs不同，DStreams的默认持久性级别将数据序列化在内存中。这将在性能调优一节中进一步讨论。有关不同持久性级别的更多信息可以在 [Spark Programming Guide][Spark] 中找到。

## Checkpointing ##

流应用程序必须全天候运行，因此必须对与应用程序逻辑无关的故障具有弹性(例如，系统故障、JVM崩溃等)。为了实现这一点，Spark 流需要将足够的信息检查点到容错存储系统，以便能够从故障中恢复。有两种类型的数据是检查点的。

*  元数据：将定义流计算的信息保存到容错存储(如HDFS)。这用于从运行流应用程序的驱动程序的节点的故障中恢复(稍后将详细讨论)。元数据包括：
    
     *  *Configuration* \- 用于创建流应用程序的配置。
     *  *DStream operations* \- 定义流应用程序的一组DStream操作。
     *  *Incomplete batches* \- 任务排队但尚未完成的批次。
 *  数据：将生成的RDDs保存到可靠的存储中。在跨多个批组合数据的一些有状态转换中，这是必需的。在这种转换中，生成的RDDs依赖于前一批的RDDs，这导致依赖链的长度随时间不断增加。为了避免这种恢复时间的无界增长(与依赖项链成比例)，有状态转换的中间rds会定期检查可靠存储(如HDFS)，以切断依赖项链。

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

### When to enable Checkpointing ###

必须启用检查点的应用程序有下列任何一项要求：

*  使用有状态转换 — 如果在应用程序中使用 updateStateByKey 或 reduceByKeyAndWindow(带有逆函数)，那么必须提供检查点目录来允许定期的 RDD 检查点。
 *  从运行应用程序的驱动程序的故障中恢复 — 使用元数据检查点来恢复进度信息。

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

### How to configure Checkpointing ###

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

*  当程序第一次启动时，它将创建一个新的 StreamingContext，设置所有的流，然后调用 start()。
 *  当程序在失败后重新启动时，它将从检查点目录中的检查点数据重新创建一个 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][]。

除了使用 getOrCreate 之外，还需要确保驱动程序进程在失败时自动重新启动。这只能由用于运行应用程序的部署基础设施来完成。

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

## Accumulators, Broadcast Variables, and Checkpoints ##

无法从 spark 流中的检查点恢复累加器和广播变量。如果启用了检查点并同时使用累加器或广播变量，则必须为累加器和广播变量创建延迟实例化的单例实例，以便在驱动程序失败重新启动后重新实例化它们。如下面的例子所示。

object WordBlacklist {
    
      @volatile private var instance: Broadcast[Seq[String]] = null
    
      def getInstance(sc: SparkContext): Broadcast[Seq[String]] = {
        if (instance == null) {
          synchronized {
            if (instance == null) {
              val wordBlacklist = Seq("a", "b", "c")
              instance = sc.broadcast(wordBlacklist)
            }
          }
        }
        instance
      }
    }
    
    object DroppedWordsCounter {
    
      @volatile private var instance: LongAccumulator = null
    
      def getInstance(sc: SparkContext): LongAccumulator = {
        if (instance == null) {
          synchronized {
            if (instance == null) {
              instance = sc.longAccumulator("WordsInBlacklistCounter")
            }
          }
        }
        instance
      }
    }
    
    wordCounts.foreachRDD { (rdd: RDD[(String, Int)], time: Time) =>
      // Get or register the blacklist Broadcast
      val blacklist = WordBlacklist.getInstance(rdd.sparkContext)
      // Get or register the droppedWordsCounter Accumulator
      val droppedWordsCounter = DroppedWordsCounter.getInstance(rdd.sparkContext)
      // Use blacklist to drop words and use droppedWordsCounter to count them
      val counts = rdd.filter { case (word, count) =>
        if (blacklist.value.contains(word)) {
          droppedWordsCounter.add(count)
          false
        } else {
          true
        }
      }.collect().mkString("[", ", ", "]")
      val output = "Counts at time " + time + " " + counts
    })

## Deploying Applications ##

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

### Requirements ###

要运行Spark流应用程序，您需要具备以下条件。

*  有集群管理器的集群——这是任何Spark应用程序的一般需求，并在部署指南中详细讨论。
 *  打包应用程序 JAR——必须将流应用程序编译到一个JAR中。如果您使用 Spark -submit 来启动应用程序，那么您将不需要在JAR中提供Spark和Spark流。但是，如果您的应用程序使用高级的源代码(例如Kafka、Flume)，那么您必须将它们链接到的额外工件及其依赖项打包到用于部署应用程序的JAR中。例如，使用 KafkaUtils 的应用程序必须在应用程序JAR中包含 spark- streamingkafka -0-8\_2.11及其所有传递依赖项。
 *  为执行器配置足够的内存——因为接收到的数据必须存储在内存中，所以必须为执行器配置足够的内存来保存接收到的数据。注意，如果您正在执行10分钟的窗口操作，则系统必须在内存中保留至少10分钟的数据。因此，应用程序的内存需求取决于它所使用的操作。
 *  配置检查点——如果流应用程序需要它，那么必须将 Hadoop API 兼容容错存储中的一个目录(如HDFS、S3等)配置为检查点目录，并以检查点信息可用于故障恢复的方式编写流应用程序。有关更多细节，请参见检查点部分。
 *  配置应用程序驱动程序的自动重启——为了从驱动程序失败中自动恢复，用于运行流应用程序的部署基础设施必须监视驱动程序进程，并在驱动程序失败时重新启动驱动程序。不同的集群管理器有不同的工具来实现这一点。
    
     *  Spark单机——可以提交一个 Spark 应用程序驱动程序在 Spark 单机集群中运行，也就是说，应用程序驱动程序本身在一个工作节点上运行。此外，可以指示独立集群管理器监视驱动程序，并在驱动程序由于非零退出码或由于运行驱动程序的节点失败而失败时重新启动它。
     *  YARN——YARN支持类似的自动重新启动应用程序的机制。
     *  Mesos -马拉松已经被用来实现这与Mesos。
 *  配置写前日志——从 Spark 1.2开始，我们就引入了写前日志，以实现强大的容错保证。如果启用，则从接收器接收到的所有数据都将写入配置检查点目录中的写前日志。这可以防止在驱动程序恢复时丢失数据，从而确保零数据丢失(在容错语义一节中详细讨论)。这可以通过设置配置参数 spark.stream .receiver. writeaheadlog 为true来启用。然而，这些更强的语义可能会以单个接收器的接收吞吐量为代价。这可以通过并行运行更多的接收器来纠正，以增加总吞吐量。此外，**建议在启用写前日志时禁用Spark中接收数据的复制**，因为日志已经存储在复制的存储系统中。这可以通过将输入流的存储级别设置为 StorageLevel.MEMORY\_AND\_DISK\_SER 来实现。**在使用S3(或任何不支持刷新的文件系统)写前日志时，请记住启用 `spark.streaming.driver.writeAheadLog.closeFileAfterWrite` and `spark.streaming.receiver.writeAheadLog.closeFileAfterWrite.`**有关详细信息，请参阅 [Spark Configuration Guide][] 。请注意，当启用I/O加密时，Spark不会加密写到写前日志的数据。如果需要对写前日志数据进行加密，则应该将其存储在本地支持加密的文件系统中。
 *  设置最大接收速率—如果集群资源不够大，流应用程序无法像接收数据那样快速处理数据，则可以通过设置记录/秒的最大速率限制来限制接收方的速率。在Spark 1.5中，我们引入了一个称为回压的特性，它消除了设置速率限制的需要，因为Spark流会自动计算速率限制，并在处理条件发生变化时动态调整它们。通过设置配置参数 spark.stream .backpressure 为 true 可以启用此背压。 之前的版本通过 `spark.streaming.receiver.maxRate` for receivers and `spark.streaming.kafka.maxRatePerPartition` 这两个参数控制接受速率。

### Upgrading Application Code ###

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

*  升级后的Spark流应用程序将与现有应用程序并行启动和运行。一旦新服务器(接收与旧服务器相同的数据)被预热并准备好进入黄金时间，旧服务器就可以被关闭。请注意，对于支持将数据发送到两个目的地(即、早期和升级的应用程序)。
 *  现有的应用程序被优雅地关闭(有关优雅的关闭选项，请参阅 StreamingContext.stop(…)或 JavaStreamingContext.stop(…))，以确保在关闭之前已经接收到的数据被完全处理。然后可以启动升级后的应用程序，它将从前面的应用程序停止的地方开始处理。请注意，这只能在支持源端缓冲的输入源(如Kafka和Flume)中完成，因为需要在前一个应用程序宕机而升级的应用程序尚未启动时对数据进行缓冲。并且无法从升级前代码的早期检查点信息重新启动。检查点信息本质上包含序列化的 Scala/Java/Python 对象，尝试使用新的、修改过的类来反序列化对象可能会导致错误。在这种情况下，可以使用不同的检查点目录启动升级后的应用程序，也可以删除以前的检查点目录

## Monitoring Applications ##

除了 Spark 的监视功能之外，还有其他特定于 Spark 流的功能。当使用 StreamingContext 时，Spark web UI 会显示一个附加的流选项卡，其中显示关于正在运行的接收方(接收方是否活动、接收到的记录数量、接收方错误等)和完成的批(批处理时间、队列延迟等)的统计信息。这可以用来监视流应用程序的进度。

web UI中的以下两个指标特别重要：

*  处理时间——处理每批数据的时间。
 *  调度延迟——批处理在队列中等待前一批处理完成的时间。

如果批处理时间始终大于批处理间隔和/或队列延迟不断增加，则表明系统无法像生成批处理那样快速地处理它们，并且正在落后。在这种情况下，可以考虑减少批处理时间。

Spark 流程序的进程也可以使用 StreamingListener 接口进行监视，该接口允许您获得接收方状态和处理时间。请注意，这是一个开发人员API，它可能会得到改进(即，更多信息报道)在未来。

# Performance Tuning #

要获得集群上 Spark 流应用程序的最佳性能，需要进行一些调优。本节解释一些参数和配置，可以对它们进行调优以提高应用程序的性能。在高层次上，你需要考虑两件事：

*  通过有效地使用集群资源，减少每批数据的处理时间。
 *  设置正确的批大小，以便能够在接收数据时快速处理这些数据(即，数据处理与数据摄入同步)。

## Reducing the Batch Processing Times ##

在Spark中可以进行许多优化，以最小化每个批的处理时间。这些在 [Tuning Spark 调优][Tuning Spark] 中有详细的讨论。本节重点介绍一些最重要的问题。

### Level of Parallelism in Data Receiving ###

通过网络接收数据(如Kafka、Flume、socket等)需要将数据反序列化并存储在 Spark 中。如果数据接收成为系统中的瓶颈，那么可以考虑将数据接收并行化。请注意，每个输入 DStream 创建一个接收方(运行在工作机器上)，接收一个数据流。

因此，通过创建多个输入数据流并配置它们以从源接收不同的数据流分区，可以实现接收多个数据流。例如，一个接收两个数据主题的 Kafka 输入 DStream 可以分成两个 Kafka 输入流，每个 Kafka 输入流只接收一个主题。这将运行两个接收器，允许并行接收数据，从而提高总体吞吐量。这些多个 DStreams 可以联合在一起创建单个 DStream。然后，应用于单个输入 DStream 上的转换可以应用于统一的流。这是这样做的。

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

另一个需要考虑的参数是接收机的块间隔，它是由配置参数 spark.stream . blockinterval 决定的。对于大多数接收方来说，接收到的数据在存储到 Spark 内存之前会被合并成数据块。每个批处理中的块的数量决定了在类映射转换中用于处理接收数据的任务的数量。每批每个接收方的任务数量大约是(批处理间隔/块间隔)。**例如，200 ms的块间隔将每2秒创建10个任务**。如果任务的数量过低(即少于每台机器的内核数量)，那么它将是低效的，因为所有可用的内核都不会被用来处理数据。若要增加给定批处理间隔的任务数量，请减少块间隔。但是，**建议的最小块间隔值为50 ms左右**，低于这个值可能会导致任务启动开销出现问题。

使用多个输入流/接收器接收数据的另一种方法是显式地重新划分输入数据流(使用 inputStream)。重新分区(<分区数量>))。在进一步处理之前，它将接收到的数据批分布到集群中指定数量的机器上。

### Level of Parallelism in Data Processing ###

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

### Data Serialization ###

通过调整序列化格式，可以减少数据序列化的开销。对于流，有两种类型的数据正在被序列化。

*  **Input data：**默认情况下，通过接收者接收到的输入数据通过 StorageLevel.MEMORY\_AND\_DISK\_SER\_2 存储在执行器的内存中。也就是说，将数据序列化为字节以减少 GC 开销，并复制数据以容忍执行程序失败。此外，数据首先保存在内存中，只有当内存不足以容纳流计算所需的所有输入数据时，才会溢出到磁盘。这种序列化显然有开销——接收方必须反序列化接收到的数据，然后使用 Spark 的序列化格式重新序列化它。
 *  **Persisted RDDs generated by Streaming Operations：**通过流计算生成的 RDDs 可以持久化到内存中。例如，窗口操作将数据保存在内存中，因为它们将被多次处理。然而，不像Spark 核心默认的 StorageLevel。通过流计算生成的持久化的 RDDs 使用 StorageLevel 持久化。默认是MEMORY\_ONLY\_SER(即序列化)，以最小化GC开销。

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

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

### Task Launching Overheads ###

如果每秒启动的任务数量很高(比如每秒50个或更多)，那么向从属服务器发送任务的开销可能会很大，并且很难实现次秒延迟。可以通过以下改变来减少开销：

*  **Execution mode：**在独立模式或粗粒度Mesos模式下运行Spark比细粒度Mesos模式的任务启动时间更长。详情请参阅[Mesos上的运行指南][Mesos]。

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

## Setting the Right Batch Interval ##

要使运行在集群上的 Spark 流应用程序稳定，系统应该能够在接收数据时快速处理数据。换句话说，处理批量数据的速度应该与生成数据的速度一样快。通过监视流 web UI 中的处理时间可以发现应用程序是否如此，在流 web UI中，**批处理时间应该小于批处理间隔。**

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

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

## Memory Tuning ##

调优 Spark 应用程序的内存使用和 GC 行为在[调优指南][Tuning Spark]中有详细的讨论。强烈建议你读一读。在本节中，我们将讨论一些特定于 Spark 流应用程序上下文的调优参数。

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

一般来说，由于通过接收器接收到的数据是用 StorageLevel.MEMORY\_AND\_DISK\_SER\_2 存储的。不适合内存的数据将溢出到磁盘。这可能会降低流应用程序的性能，因此建议根据流应用程序的需要提供足够的内存。**最好尝试在小范围内查看内存使用情况并进行相应的估计。**

内存调优的另一个方面是垃圾收集。对于需要低延迟的流应用程序，JV M垃圾收集导致的大量暂停是不可取的。

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

*  **Persistence Level of DStreams：**正如前面在数据序列化一节中提到的，输入数据和 RDDs 在默认情况下是作为序列化字节持久化的。与反序列化持久性相比，这减少了内存使用和 GC 开销。启用 Kryo 序列化进一步减少了序列化的大小和内存使用。通过压缩(参见Spark配置 Spark .rdd.compress )可以进一步减少内存使用量，但要以 CPU 时间为代价。
 *  **Clearing old data**: 默认情况下，由 DStream 转换生成的所有输入数据和持久的 RDDs 将被自动清除。Spark 流根据所使用的转换决定何时清除数据。例如，如果您使用的是10分钟的窗口操作，那么 Spark 流将保留最后10分钟的数据，并主动丢弃旧的数据。通过设置 streamingContext.remember，可以更长时间地保留数据(例如交互式地查询旧数据)。
 *  **CMS Garbage Collector**: 强烈建议使用并发标记-清除 GC，以保持与 GC 相关的暂停始终较低。尽管众所周知并发GC会降低系统的总体处理吞吐量，但**仍然建议使用它来实现更一致的批处理时间**。确保在驱动程序(在 Spark -submit 中使用—driver-java-options)和执行器(使用Spark配置Spark .executor. extrajavaoptions)上都设置了CMS GC。
 *  **Other tips**: 为了进一步减少GC开销，这里还有一些技巧可以尝试。
    
     *  使用OFF\_HEAP存储级别持久化RDDs. See more detail in the [Spark Programming Guide][].
     *  使用更多的执行器和更小的堆大小。这将减少每个JVM堆中的GC压力。

**Important points to remember:**

*  DStream 与单个接收器相关联。为了获得读并行性，需要创建多个接收器，即多个 DStreams。接收器在执行程序中运行。它占据一个核。确保在接收槽被预定后有足够的 core 用于处理，即 spark.core.max 应该考虑接收槽。接收者以循环方式分配给执行者。
 *  当从流源接收数据时，receiver 创建数据块。每隔一毫秒就会产生一个新的数据块。在 batchInterval 期间创建N个数据块，其中N = batchInterval/blockInterval。这些块由当前执行程序的块管理器分发给其他执行程序的块管理器。之后，运行在驱动程序上的网络输入跟踪器将被告知进一步处理的块位置。
 *  在 batchInterval 期间创建的块的驱动程序上创建了一个RDD。batchInterval 期间生成的块是 RDD 的分区。每个分区都是spark 中的一个任务。blockInterval== batchinterval 将意味着创建一个单独的分区，并且可能在本地处理它。
 *  块上的映射任务在执行器(一个接收块，另一个在复制块)中处理，不管块的间隔如何，除非出现非本地调度。拥有更大的块间隔意味着更大的块。`spark.locality.wait `增加了在本地节点上处理一个块的机会。需要在这两个参数之间找到平衡，以确保在本地处理较大的块。
 *  您可以通过调用 inputDstream.repartition(n) 来定义分区的数量，而不是依赖于 batchInterval 和 blockInterval。这将随机重组 RDD 中的数据，以创建n个分区。是的，为了更好的并行性。不过代价是洗牌。RDD 的处理是由驱动程序的 jobscheduler 作为作业调度的。在给定的时间点上，只有一个作业是活动的。因此，如果一个作业正在执行，其他作业将排队。
 *  如果您有两个 dstreams，则会形成两个 RDDs，并创建两个作业，这些作业将一个接一个地安排。为了避免这种情况，您可以联合两个 dstreams。这将确保 dstreams 的两个rds形成一个 unionRDD。然后，这 个unionRDD 被视为单个作业。但是，RDDs 的分区不受影响。
 *  如果批处理时间大于 batchinterval，那么显然接收方的内存将开始填满，并最终抛出异常(很可能是BlockNotFoundException)。目前没有办法暂停接收器。使用 SparkConf 配置 spark.stream .receiver.maxRate，接收速率是有限的。

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

# Fault-tolerance Semantics #

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

## Background ##

为了理解 Spark 流提供的语义，让我们记住 Spark 的 RDDs 的基本容错语义。

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

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

*  接收和复制的数据——由于单个工作节点的副本存在于其他节点上，所以该数据在单个工作节点失败后仍然存在。
 *  接收到但为复制而缓冲的数据——由于这不是复制，因此恢复此数据的惟一方法是从源获取它。

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

*  工作节点失败——运行 executor 的任何工作节点都可能失败，这些节点上的所有内存数据都将丢失。如果任何接收器在失败的节点上运行，那么它们的缓冲数据将丢失。
 *  驱动节点失败——如果运行 Spark 流应用程序的驱动节点失败，那么很明显，SparkContext 丢失了，所有执行器及其内存中的数据都丢失了。

有了这些基础知识，我们就可以理解Spark流的容错语义。

## 容错Definitions ##

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

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

## Basic Semantics ##

一般来说，在任何流处理系统中，处理数据有三个步骤。

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

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

1.  接收数据:不同的输入源提供不同的保证。这将在下一小节中详细讨论。
2.  数据转换:由于 RDDs 提供的保证，所有接收到的数据都将被精确地处理一次。即使存在故障，只要接收的输入数据是可访问的，最终转换的 RDDs 将始终具有相同的内容。
3.  输出数据:默认情况下，输出操作至少确保一次语义，因为它取决于输出操作的类型(幂等性，或非幂等性)和下游系统的语义(支持或不支持事务)。但是用户可以实现自己的事务机制来实现精确的一次语义。本节后面将更详细地讨论这一点。

## Semantics of Received Data ##

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

### With Files ###

如果所有的输入数据都已经存在于一个容错的文件系统(如HDFS)中，那么 Spark 流始终可以从任何故障中恢复并处理所有的数据。这提供了严格的一次语义，这意味着所有的数据都将被精确地处理一次，不管什么失败。

### With Receiver-based Sources ###

对于基于接收器的输入源，容错语义依赖于故障场景和接收器的类型。如前所述，有两种类型的接收器：

1.  可靠的接收器-这些接收器确认可靠的来源后，才确保收到的数据已被复制。如果这样的接收器失败，源将不会收到缓冲(未复制)数据的确认。因此，如果重新启动接收方，则源将重新发送数据，并且不会因为失败而丢失任何数据。
2.  不可靠的接收器——这类接收器不发送确认信息，因此，当它们由于工人或驱动程序故障而失败时，可能会丢失数据。

根据所使用的接收器类型，我们可以实现以下语义。如果工作节点失败，那么可靠的接收器就不会丢失数据。对于不可靠的接收器，接收到但没有复制的数据可能会丢失。如果驱动节点失败，那么除了这些丢失之外，所有在内存中接收和复制的过去数据都将丢失。这将影响有状态转换的结果。

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

下表总结了故障下的语义:

<table> 
 <tbody> 
 <tr> 
 <th>Deployment Scenario</th> 
 <th>Worker Failure</th> 
 <th>Driver Failure</th> 
 </tr> 
 <tr> 
 <td>Spark 1.1 or earlier, OR Spark 1.2 or later without write ahead logs</td> 
 <td>缓冲数据丢失与不可靠的接收器 零数据损失与可靠的接收器 至少一次语义</td> 
 <td>缓冲数据丢失与不可靠的接收器 过去的数据丢失与所有的接收器 未定义的语义</td> 
 </tr> 
 <tr> 
 <td>Spark 1.2 or later with write ahead logs</td> 
 <td>零数据损失与可靠的接收器 至少一次语义</td> 
 <td>零数据丢失与可靠的接收器和文件 至少一次语义</td> 
 </tr> 
 </tbody> 
</table>

### With Kafka Direct API ###

在Spark 1.3中，我们引入了一个新的 Kafka 直接 API，它可以保证所有的 Kafka 数据被 Spark 流一次接收。与此同时，如果您实现精确一次输出操作，则可以实现端到端的精确一次输出保证。[Kafka集成指南][Kafka]进一步讨论了这种方法。

## Semantics of output operations ##

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

*  幂等更新:多次尝试总是写入相同的数据。例如，saveAs\*\*\*文件总是将相同的数据写入生成的文件。
 *  事务性更新:所有的更新都是通过事务方式进行的，这样就可以原子性地进行一次更新。一种方法是这样的。
    
     *  使用批处理时间(在 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
      }
    }

原文地址：[http://spark.apache.org/docs/2.1.1/streaming-programming-guide.html\#initializing-streamingcontext][http_spark.apache.org_docs_2.1.1_streaming-programming-guide.html_initializing-streamingcontext]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200306115031428.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI=,size_16,color_FFFFFF,t_70
[20200306115055525.png]: https://img-blog.csdnimg.cn/20200306115055525.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200306120000522.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI=,size_16,color_FFFFFF,t_70
[Spark]: https://blog.csdn.net/zpf_940810653842/article/details/103350992
[202003061758450.png]: https://img-blog.csdnimg.cn/202003061758450.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200306180216699.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI=,size_16,color_FFFFFF,t_70
[https_blog.csdn.net_zpf_940810653842_article_details_103372744_spark-properties]: https://blog.csdn.net/zpf_940810653842/article/details/103372744#spark-properties
[Spark Streaming Custom Receivers]: https://blog.csdn.net/zpf_940810653842/article/details/104701498
[http_spark.apache.org_docs_2.1.1_streaming-kafka-integration.html]: http://spark.apache.org/docs/2.1.1/streaming-kafka-integration.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200307110901351.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pwZl85NDA4MTA2NTM4NDI=,size_16,color_FFFFFF,t_70
[DStream]: http://spark.apache.org/docs/2.1.1/api/scala/index.html#org.apache.spark.streaming.dstream.DStream
[PairDStreamFunctions]: http://spark.apache.org/docs/2.1.1/api/scala/index.html#org.apache.spark.streaming.dstream.PairDStreamFunctions
[JavaDStream]: http://spark.apache.org/docs/2.1.1/api/java/index.html?org/apache/spark/streaming/api/java/JavaDStream.html
[JavaPairDStream]: http://spark.apache.org/docs/2.1.1/api/java/index.html?org/apache/spark/streaming/api/java/JavaPairDStream.html
[DStream 1]: http://spark.apache.org/docs/2.1.1/api/python/pyspark.streaming.html#pyspark.streaming.DStream
[Spark SQL_ DataFrames and Datasets Guide]: https://blog.csdn.net/zpf_940810653842/article/details/104716352
[RecoverableNetworkWordCount]: https://github.com/apache/spark/tree/master/examples/src/main/scala/org/apache/spark/examples/streaming/RecoverableNetworkWordCount.scala
[Spark Configuration Guide]: https://blog.csdn.net/zpf_940810653842/article/details/103372744
[Tuning Spark]: https://blog.csdn.net/zpf_940810653842/article/details/103409760
[Mesos]: http://spark.apache.org/docs/2.1.1/running-on-mesos.html
[Spark Programming Guide]: http://spark.apache.org/docs/2.1.1/programming-guide.html#rdd-persistence
[Kafka]: https://blog.csdn.net/zpf_940810653842/article/details/104675468
[http_spark.apache.org_docs_2.1.1_streaming-programming-guide.html_initializing-streamingcontext]: http://spark.apache.org/docs/2.1.1/streaming-programming-guide.html#initializing-streamingcontext