Spark Streaming Programming Guide
Spark Streaming编程指南

• Overview  概述
• A Quick Example  一个简单的例子
• Basic Concepts  基本概念
  • Linking  连接
  • Initializing StreamingContext
    正在初始化StreamingContext
  • Discretized Streams (DStreams)
    离散流（DStreams）
  • Input DStreams and Receivers
    输入数据流和接收器
  • Transformations on DStreams
    DStreams上的转换
  • Output Operations on DStreams
    DStreams上的输出操作
  • DataFrame and SQL Operations
    DataFrame和SQL操作
  • MLlib Operations  MLlib操作
  • Caching / Persistence  缓存/持久性
  • Checkpointing  检查点
  • Accumulators, Broadcast Variables, and Checkpoints
    累加器、广播变量和检查点
  • Deploying Applications  部署应用程序
  • Monitoring Applications  监测应用
• Performance Tuning  性能调优
  • Reducing the Batch Processing Times
    减少批处理时间
  • Setting the Right Batch Interval
    设置正确的批处理间隔
  • Memory Tuning  内存调优
• Fault-tolerance Semantics
  容错语义
• Where to Go from Here
  从这里往哪里走

Overview 概述

Spark Streaming is an extension of the core Spark API that enables scalable, high-throughput, fault-tolerant stream processing of live data streams. Data can be ingested from many sources like Kafka, Kinesis, or TCP sockets, and can be processed using complex algorithms expressed with high-level functions like map, reduce, join and window. Finally, processed data can be pushed out to filesystems, databases, and live dashboards. In fact, you can apply Spark’s machine learning and graph processing algorithms on data streams.
Spark Streaming是核心Spark API的扩展，支持实时数据流的可扩展、高吞吐量、容错流处理。数据可以从许多源（如Kafka、Kinesis或TCP套接字）中获取，并且可以使用复杂的算法（如 map 、 reduce 、 join 和 window 等高级函数）进行处理。最后，可以将处理后的数据推送到文件系统、数据库和实时仪表板。事实上，您可以在数据流上应用Spark的机器学习和图形处理算法。

Internally, it works as follows. Spark Streaming receives live input data streams and divides the data into batches, which are then processed by the Spark engine to generate the final stream of results in batches.
在内部，它的工作原理如下。Spark Streaming接收实时输入数据流，并将数据分成批次，然后由Spark引擎处理以批量生成最终结果流。

Spark Streaming provides a high-level abstraction called discretized stream or DStream, which represents a continuous stream of data. DStreams can be created either from input data streams from sources such as Kafka, and Kinesis, or by applying high-level operations on other DStreams. Internally, a DStream is represented as a sequence of RDDs.
Spark Streaming提供了一种称为离散流或DStream的高级抽象，它表示连续的数据流。DStreams可以从Kafka和Kinesis等源的输入数据流创建，也可以通过对其他DStreams应用高级操作来创建。在内部，DStream被表示为RDD序列。

This guide shows you how to start writing Spark Streaming programs with DStreams. You can write Spark Streaming programs in Scala, Java or Python (introduced in Spark 1.2), all of which are presented in this guide. You will find tabs throughout this guide that let you choose between code snippets of different languages.
本指南向您展示如何开始使用DStreams编写Spark Streaming程序。您可以使用Scala、Java或Python（在Spark 1.2中引入）编写Spark Streaming程序，所有这些都将在本指南中介绍。您将在本指南中找到选项卡，可以在不同语言的代码段之间进行选择。

Note: There are a few APIs that are either different or not available in Python. Throughout this guide, you will find the tag Python API highlighting these differences.
注意：有一些API在Python中是不同的或不可用的。在本指南中，您会发现标记Python API突出显示了这些差异。

A Quick Example 一个简单的例子

Before we go into the details of how to write your own Spark Streaming program, let’s take a quick look at what a simple Spark Streaming program looks like. Let’s say we want to count the number of words in text data received from a data server listening on a TCP socket. All you need to do is as follows.
在详细介绍如何编写自己的Spark Streaming程序之前，让我们快速了解一下简单的Spark Streaming程序是什么样子的。假设我们想要计算从侦听TCP套接字的数据服务器接收的文本数据中的单词数。所有你需要做的是如下。

• Scala
• Java
• Python

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

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

If you have already downloaded and built Spark, you can run this example as follows. You will first need to run Netcat (a small utility found in most Unix-like systems) as a data server by using
如果您已经下载并构建了Spark，则可以按如下方式运行此示例。您首先需要使用以下命令将Netcat（在大多数类Unix系统中可以找到的一个小型实用程序）作为数据服务器运行：

$ nc -lk 9999

Then, in a different terminal, you can start the example by using
然后，在另一个终端中，您可以使用

• Scala
• Java
• Python

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

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

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

Then, any lines typed in the terminal running the netcat server will be counted and printed on screen every second. It will look something like the following.
然后，在运行netcat服务器的终端中键入的任何行将每秒计数并打印在屏幕上。它看起来像下面这样。

# TERMINAL 1:
# Running Netcat

$ nc -lk 9999

hello world



...

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

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

Basic Concepts 基本概念

Next, we move beyond the simple example and elaborate on the basics of Spark Streaming.
接下来，我们将超越简单的示例，详细介绍Spark Streaming的基础知识。

Linking 连接

Similar to Spark, Spark Streaming is available through Maven Central. To write your own Spark Streaming program, you will have to add the following dependency to your SBT or Maven project.
与Spark类似，Spark Streaming可以通过Maven Central获得。要编写自己的Spark Streaming程序，您必须将以下依赖项添加到您的SBT或Maven项目中。

• Maven
• SBT

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

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

For ingesting data from sources like Kafka and Kinesis that are not present in the Spark Streaming core API, you will have to add the corresponding artifact spark-streaming-xyz_2.12 to the dependencies. For example, some of the common ones are as follows.
对于来自Kafka和Kinesis等源的ingredient数据，如果这些数据在Spark Streaming核心API中不存在，则必须将相应的工件 spark-streaming-xyz_2.12 添加到依赖项中。例如，一些常见的如下。

For an up-to-date list, please refer to the Maven repository for the full list of supported sources and artifacts.
有关最新列表，请参阅Maven存储库，以获取支持的源代码和工件的完整列表。

Initializing StreamingContext
正在初始化StreamingContext

To initialize a Spark Streaming program, a StreamingContext object has to be created which is the main entry point of all Spark Streaming functionality.
要初始化Spark Streaming程序，必须创建StreamingContext对象，它是所有Spark Streaming功能的主要入口点。

• Scala
• Java
• Python

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

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

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

After a context is defined, you have to do the following.
在定义了上下文之后，您必须执行以下操作。

1. Define the input sources by creating input DStreams.
   通过创建输入DStreams定义输入源。
2. Define the streaming computations by applying transformation and output operations to DStreams.
   通过对DStreams应用转换和输出操作来定义流计算。
3. Start receiving data and processing it using streamingContext.start().
   开始接收数据并使用 streamingContext.start() 进行处理。
4. Wait for the processing to be stopped (manually or due to any error) using streamingContext.awaitTermination().
   使用 streamingContext.awaitTermination() 等待处理停止（手动或由于任何错误）。
5. The processing can be manually stopped using streamingContext.stop().
   可以使用 streamingContext.stop() 手动停止处理。

Points to remember: 要记住的要点：

• Once a context has been started, no new streaming computations can be set up or added to it.
  一旦启动了上下文，就不能设置或添加新的流计算。
• Once a context has been stopped, it cannot be restarted.
  上下文一旦停止，就无法重新启动。
• Only one StreamingContext can be active in a JVM at the same time.
  一个JVM中只能同时有一个StreamingContext处于活动状态。
• stop() on StreamingContext also stops the SparkContext. To stop only the StreamingContext, set the optional parameter of stop() called stopSparkContext to false.
  StreamingContext上的stop（）也会停止SparkContext。要仅停止StreamingContext，请将 stop() 的可选参数 stopSparkContext 设置为false。
• A SparkContext can be re-used to create multiple StreamingContexts, as long as the previous StreamingContext is stopped (without stopping the SparkContext) before the next StreamingContext is created.
  一个SparkContext可以被重用来创建多个StreamingContext，只要在创建下一个StreamingContext之前停止前一个StreamingContext（不停止SparkContext）。

Discretized Streams (DStreams)
离散流（DStreams）

Discretized Stream or DStream is the basic abstraction provided by Spark Streaming. It represents a continuous stream of data, either the input data stream received from source, or the processed data stream generated by transforming the input stream. Internally, a DStream is represented by a continuous series of RDDs, which is Spark’s abstraction of an immutable, distributed dataset (see Spark Programming Guide for more details). Each RDD in a DStream contains data from a certain interval, as shown in the following figure.
离散流或DStream是Spark Streaming提供的基本抽象。它表示一个连续的数据流，或者是从源接收的输入数据流，或者是通过转换输入流生成的处理后的数据流。在内部，DStream由一系列连续的RDD表示，RDD是Spark对不可变的分布式数据集的抽象（更多细节请参见Spark编程指南）。DStream中的每个RDD都包含来自某个时间间隔的数据，如下图所示。

Any operation applied on a DStream translates to operations on the underlying RDDs. For example, in the earlier example of converting a stream of lines to words, the flatMap operation is applied on each RDD in the lines DStream to generate the RDDs of the words DStream. This is shown in the following figure.
应用于DStream的任何操作都转换为底层RDD上的操作。例如，在将行流转换为字的较早示例中，对 lines DStream中的每个RDD应用 flatMap 操作以生成 words DStream的RDD。如下图所示。

These underlying RDD transformations are computed by the Spark engine. The DStream operations hide most of these details and provide the developer with a higher-level API for convenience. These operations are discussed in detail in later sections.
这些底层的RDD转换由Spark引擎计算。DStream操作隐藏了大部分这些细节，并为开发人员提供了更高级别的API以方便。这些操作将在后面的章节中详细讨论。

Input DStreams and Receivers
输入数据流和接收器

Input DStreams are DStreams representing the stream of input data received from streaming sources. In the quick example, lines was an input DStream as it represented the stream of data received from the netcat server. Every input DStream (except file stream, discussed later in this section) is associated with a Receiver (Scala doc, Java doc) object which receives the data from a source and stores it in Spark’s memory for processing.
输入数据流是表示从流传输源接收的输入数据流的数据流。在快速示例中， lines 是一个输入DStream，因为它表示从netcat服务器接收的数据流。每个输入DStream（除了文件流，在本节后面讨论）都与一个Receiver（Scala doc，Java doc）对象相关联，该对象从源接收数据并将其存储在Spark的内存中进行处理。

Spark Streaming provides two categories of built-in streaming sources.
Spark Streaming提供两类内置流媒体源。

• Basic sources: Sources directly available in the StreamingContext API. Examples: file systems, and socket connections.
  基本源：StreamingContext API中直接提供的源。例如：文件系统和套接字连接。
• Advanced sources: Sources like Kafka, Kinesis, etc. are available through extra utility classes. These require linking against extra dependencies as discussed in the linking section.
  高级源代码：像Kafka，Kinesis等源代码可以通过额外的实用程序类获得。这些需要针对额外的依赖项进行链接，如链接部分所述。

We are going to discuss some of the sources present in each category later in this section.
我们将在本节后面讨论每个类别中的一些来源。

Note that, if you want to receive multiple streams of data in parallel in your streaming application, you can create multiple input DStreams (discussed further in the Performance Tuning section). This will create multiple receivers which will simultaneously receive multiple data streams. But note that a Spark worker/executor is a long-running task, hence it occupies one of the cores allocated to the Spark Streaming application. Therefore, it is important to remember that a Spark Streaming application needs to be allocated enough cores (or threads, if running locally) to process the received data, as well as to run the receiver(s).
请注意，如果您希望在流应用程序中并行接收多个数据流，则可以创建多个输入DStream（在性能调优部分中进一步讨论）。这将创建多个接收器，这些接收器将同时接收多个数据流。但是请注意，Spark worker/executor是一个长时间运行的任务，因此它占用了分配给Spark Streaming应用程序的一个核心。因此，重要的是要记住，Spark Streaming应用程序需要分配足够的核心（或线程，如果在本地运行）来处理接收到的数据，以及运行接收器。

Points to remember 要记住的要点

• When running a Spark Streaming program locally, do not use “local” or “local[1]” as the master URL. Either of these means that only one thread will be used for running tasks locally. If you are using an input DStream based on a receiver (e.g. sockets, Kafka, etc.), then the single thread will be used to run the receiver, leaving no thread for processing the received data. Hence, when running locally, always use “local[n]” as the master URL, where n > number of receivers to run (see Spark Properties for information on how to set the master).
  在本地运行Spark Streaming程序时，不要使用"local"或"local [1]"作为主URL。这两种情况都意味着只有一个线程将用于在本地运行任务。如果你使用的是基于接收器的输入DStream（例如sockets、Kafka等），则该单个线程将用于运行接收器，而不留下用于处理接收到的数据的线程。因此，在本地运行时，始终使用"local [n]"作为主URL，其中n>要运行的接收器数量（有关如何设置主的信息，请参阅Spark Properties）。

• Extending the logic to running on a cluster, the number of cores allocated to the Spark Streaming application must be more than the number of receivers. Otherwise the system will receive data, but not be able to process it.
  将逻辑扩展到在集群上运行，分配给Spark Streaming应用程序的核心数量必须大于接收器的数量。否则，系统将接收数据，但无法处理它。

Basic Sources 基本来源

We have already taken a look at the ssc.socketTextStream(...) in the quick example which creates a DStream from text data received over a TCP socket connection. Besides sockets, the StreamingContext API provides methods for creating DStreams from files as input sources.
我们已经看过了快速示例中的 ssc.socketTextStream(...) ，它从通过TCP套接字连接接收的文本数据创建DStream。除了套接字之外，StreamingContext API还提供了从作为输入源的文件创建DStreams的方法。

File Streams 文件流

For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as via StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass].
对于从与HDFS API（即HDFS、S3、NFS等）兼容的任何文件系统上的文件中阅读数据，可以通过 StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass] 创建DStream。

File streams do not require running a receiver so there is no need to allocate any cores for receiving file data.
文件流不需要运行接收器，因此不需要分配任何核心来接收文件数据。

For simple text files, the easiest method is StreamingContext.textFileStream(dataDirectory).
对于简单的文本文件，最简单的方法是 StreamingContext.textFileStream(dataDirectory) 。

• Scala
• Java
• Python

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

streamingContext.textFileStream(dataDirectory)

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

streamingContext.textFileStream(dataDirectory);

streamingContext.textFileStream(dataDirectory)

How Directories are Monitored
目录如何分类

Spark Streaming will monitor the directory dataDirectory and process any files created in that directory.
Spark Streaming将监控目录 dataDirectory 并处理该目录中创建的任何文件。

• A simple directory can be monitored, such as "hdfs://namenode:8040/logs/". All files directly under such a path will be processed as they are discovered.
  可以监视一个简单的目录，例如 "hdfs://namenode:8040/logs/" 。直接位于该路径下的所有文件将在发现时进行处理。
• A POSIX glob pattern can be supplied, such as "hdfs://namenode:8040/logs/2017/*". Here, the DStream will consist of all files in the directories matching the pattern. That is: it is a pattern of directories, not of files in directories.
  可以提供POSIX glob模式，例如 "hdfs://namenode:8040/logs/2017/*" 。在这里，DStream将由目录中与模式匹配的所有文件组成。也就是说：它是目录的模式，而不是目录中的文件的模式。
• All files must be in the same data format.
  所有文件必须采用相同的数据格式。
• A file is considered part of a time period based on its modification time, not its creation time.
  根据文件的修改时间（而不是创建时间），文件被视为时间段的一部分。
• Once processed, changes to a file within the current window will not cause the file to be reread. That is: updates are ignored.
  一旦被处理，在当前窗口中对文件的更改将不会导致重新读取该文件。也就是说：更新被忽略。
• The more files under a directory, the longer it will take to scan for changes — even if no files have been modified.
  目录下的文件越多，扫描更改所需的时间就越长-即使没有文件被修改。
• If a wildcard is used to identify directories, such as "hdfs://namenode:8040/logs/2016-*", renaming an entire directory to match the path will add the directory to the list of monitored directories. Only the files in the directory whose modification time is within the current window will be included in the stream.
  如果使用一个索引来标识目录，如 "hdfs://namenode:8040/logs/2016-*" ，重命名整个目录以匹配路径将把该目录添加到受监视目录的列表中。只有修改时间在当前窗口内的目录中的文件才会包含在流中。
• Calling FileSystem.setTimes() to fix the timestamp is a way to have the file picked up in a later window, even if its contents have not changed.
  调用 FileSystem.setTimes() 来修复时间戳是一种在以后的窗口中拾取文件的方法，即使其内容没有更改。

Using Object Stores as a source of data
使用对象存储作为数据源

“Full” Filesystems such as HDFS tend to set the modification time on their files as soon as the output stream is created. When a file is opened, even before data has been completely written, it may be included in the DStream - after which updates to the file within the same window will be ignored. That is: changes may be missed, and data omitted from the stream.
“完整”文件系统（如HDFS）倾向于在创建输出流时立即在其文件上设置修改时间。当一个文件被打开时，即使在数据被完全写入之前，它也可能被包含在 DStream 中，之后在同一窗口内对文件的更新将被忽略。也就是说：更改可能会丢失，数据可能会从流中遗漏。

To guarantee that changes are picked up in a window, write the file to an unmonitored directory, then, immediately after the output stream is closed, rename it into the destination directory. Provided the renamed file appears in the scanned destination directory during the window of its creation, the new data will be picked up.
要保证在窗口中拾取更改，请将文件写入未监视的目录，然后在输出流关闭后立即将其重命名为目标目录。如果重命名的文件在其创建窗口期间出现在扫描的目标目录中，则将拾取新数据。

In contrast, Object Stores such as Amazon S3 and Azure Storage usually have slow rename operations, as the data is actually copied. Furthermore, renamed object may have the time of the rename() operation as its modification time, so may not be considered part of the window which the original create time implied they were.
相比之下，对象存储（如Amazon S3和Azure存储）的重命名操作通常很慢，因为数据实际上是复制的。此外，重命名的对象可以将 rename() 操作的时间作为其修改时间，因此可以不被认为是原始创建时间暗示它们是的窗口的一部分。

Careful testing is needed against the target object store to verify that the timestamp behavior of the store is consistent with that expected by Spark Streaming. It may be that writing directly into a destination directory is the appropriate strategy for streaming data via the chosen object store.
需要对目标对象存储进行仔细的测试，以验证存储的时间戳行为与Spark Streaming的预期一致。直接写入目标目录可能是通过所选对象存储流式传输数据的适当策略。

For more details on this topic, consult the Hadoop Filesystem Specification.
有关此主题的更多详细信息，请参阅Hadoop文件系统规范。

Streams based on Custom Receivers
基于自定义接收器的流

DStreams can be created with data streams received through custom receivers. See the Custom Receiver Guide for more details.
DStreams可以使用通过自定义接收器接收的数据流创建。有关详细信息，请参阅自定义接收器指南。

Queue of RDDs as a Stream
RDD队列作为流

For testing a Spark Streaming application with test data, one can also create a DStream based on a queue of RDDs, using streamingContext.queueStream(queueOfRDDs). Each RDD pushed into the queue will be treated as a batch of data in the DStream, and processed like a stream.
为了使用测试数据测试Spark Streaming应用程序，还可以使用 streamingContext.queueStream(queueOfRDDs) 基于RDD队列创建DStream。每个推入队列的RDD将被视为DStream中的一批数据，并像流一样处理。

For more details on streams from sockets and files, see the API documentations of the relevant functions in StreamingContext for Scala, JavaStreamingContext for Java, and StreamingContext for Python.
有关来自套接字和文件的流的更多详细信息，请参阅StreamingContext for Scala、JavaStreamingContext for Java和StreamingContext for Python中相关函数的API文档。

Advanced Sources 先进的来源

Python API As of Spark 3.1.2, out of these sources, Kafka and Kinesis are available in the Python API.
Python API从Spark 3.1.2开始，在这些源代码中，Kafka和Kinesis可以在Python API中使用。

This category of sources requires interfacing with external non-Spark libraries, some of them with complex dependencies (e.g., Kafka). Hence, to minimize issues related to version conflicts of dependencies, the functionality to create DStreams from these sources has been moved to separate libraries that can be linked to explicitly when necessary.
这类源需要与外部非Spark库接口，其中一些库具有复杂的依赖关系（例如，Kafka）。因此，为了最大限度地减少与依赖项的版本冲突相关的问题，从这些源创建DStreams的功能已被转移到单独的库中，这些库可以在必要时显式链接。

Note that these advanced sources are not available in the Spark shell, hence applications based on these advanced sources cannot be tested in the shell. If you really want to use them in the Spark shell you will have to download the corresponding Maven artifact’s JAR along with its dependencies and add it to the classpath.
请注意，这些高级源代码在Spark shell中不可用，因此基于这些高级源代码的应用程序不能在shell中测试。如果你真的想在Spark shell中使用它们，你将不得不下载相应的Maven工件的包沿着它的依赖项，并将其添加到类路径中。

Some of these advanced sources are as follows.
其中一些先进的来源如下。

• Kafka: Spark Streaming 3.1.2 is compatible with Kafka broker versions 0.10 or higher. See the Kafka Integration Guide for more details.
  Kafka：Spark Streaming 3.1.2与Kafka broker 0.10或更高版本兼容。有关详细信息，请参阅Kafka集成指南。

• Kinesis: Spark Streaming 3.1.2 is compatible with Kinesis Client Library 1.2.1. See the Kinesis Integration Guide for more details.
  Kinesis：Spark Streaming 3.1.2与Kinesis Client Library 1.2.1兼容。有关详细信息，请参阅《Kinesis集成指南》。

Custom Sources 自定义来源

Python API This is not yet supported in Python.
Python API Python中尚不支持此功能。

Input DStreams can also be created out of custom data sources. All you have to do is implement a user-defined receiver (see next section to understand what that is) that can receive data from the custom sources and push it into Spark. See the Custom Receiver Guide for details.
输入DStream也可以从自定义数据源创建。您所要做的就是实现一个用户定义的接收器（请参阅下一节以了解它是什么），它可以从自定义源接收数据并将其推送到Spark中。有关详细信息，请参阅《自定义接收器指南》。

Receiver Reliability 接收方可靠性

There can be two kinds of data sources based on their reliability. Sources (like Kafka) allow the transferred data to be acknowledged. If the system receiving data from these reliable sources acknowledges the received data correctly, it can be ensured that no data will be lost due to any kind of failure. This leads to two kinds of receivers:
根据可靠性，可以有两种数据源。源（如Kafka）允许确认传输的数据。如果从这些可靠来源接收数据的系统正确地确认接收到的数据，则可以确保不会由于任何类型的故障而丢失数据。这导致两种接收器：

1. Reliable Receiver - A reliable receiver correctly sends acknowledgment to a reliable source when the data has been received and stored in Spark with replication.
   可靠的接收者—当数据被接收并存储在Spark中时，可靠的接收者正确地向可靠的源发送确认。
2. Unreliable Receiver - An unreliable receiver does not send acknowledgment to a source. This can be used for sources that do not support acknowledgment, or even for reliable sources when one does not want or need to go into the complexity of acknowledgment.
   不可靠的接收方—不可靠的接收方不向源发送确认。这可以用于不支持确认的源，或者甚至用于不想或不需要进入确认的复杂性的可靠源。

The details of how to write a reliable receiver are discussed in the Custom Receiver Guide.
如何编写可靠的接收器的细节在自定义接收器指南中讨论。

Transformations on DStreams
DStreams上的转换

Similar to that of RDDs, transformations allow the data from the input DStream to be modified. DStreams support many of the transformations available on normal Spark RDD’s. Some of the common ones are as follows.
与RDD类似，转换允许修改来自输入DStream的数据。DStreams支持许多在普通Spark RDD上可用的转换。一些常见的如下。

A few of these transformations are worth discussing in more detail.
其中有几个转换值得更详细地讨论。

UpdateStateByKey Operation
UpdateStateByKey操作

The updateStateByKey operation allows you to maintain arbitrary state while continuously updating it with new information. To use this, you will have to do two steps.
updateStateByKey 操作允许您在使用新信息不断更新状态的同时保持任意状态。要使用它，你必须做两个步骤。

1. Define the state - The state can be an arbitrary data type.
   定义状态-状态可以是任意数据类型。
2. Define the state update function - Specify with a function how to update the state using the previous state and the new values from an input stream.
   定义状态更新函数-指定如何使用以前的状态和来自输入流的新值来更新状态。

In every batch, Spark will apply the state update function for all existing keys, regardless of whether they have new data in a batch or not. If the update function returns None then the key-value pair will be eliminated.
在每个批处理中，Spark都会对所有现有的密钥应用状态更新函数，无论它们是否在批处理中有新的数据。如果update函数返回 None ，那么键值对将被删除。

Let’s illustrate this with an example. Say you want to maintain a running count of each word seen in a text data stream. Here, the running count is the state and it is an integer. We define the update function as:
让我们用一个例子来说明这一点。假设您想要维护文本数据流中每个单词的运行计数。这里，运行计数是状态，并且是一个整数。我们将update函数定义为：

• Scala
• Java
• Python

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

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)

Note that using updateStateByKey requires the checkpoint directory to be configured, which is discussed in detail in the checkpointing section.
请注意，使用 updateStateByKey 需要配置检查点目录，这将在检查点一节中详细讨论。

Transform Operation 变换操作

The transform operation (along with its variations like transformWith) allows arbitrary RDD-to-RDD functions to be applied on a DStream. It can be used to apply any RDD operation that is not exposed in the DStream API. For example, the functionality of joining every batch in a data stream with another dataset is not directly exposed in the DStream API. However, you can easily use transform to do this. This enables very powerful possibilities. For example, one can do real-time data cleaning by joining the input data stream with precomputed spam information (maybe generated with Spark as well) and then filtering based on it.
transform 操作（沿着其变体，如 transformWith ）允许在DStream上应用任意RDD到RDD函数。它可用于应用任何未在DStream API中公开的RDD操作。例如，将数据流中的每个批处理与另一个数据集连接的功能不会直接在DStream API中公开。但是，您可以轻松地使用 transform 来完成此操作。这使得非常强大的可能性。例如，可以通过将输入数据流与预先计算的垃圾信息（也可能是Spark生成的）结合起来，然后基于它进行过滤，来进行实时数据清理。

• Scala
• Java
• Python

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

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

Note that the supplied function gets called in every batch interval. This allows you to do time-varying RDD operations, that is, RDD operations, number of partitions, broadcast variables, etc. can be changed between batches.
请注意，提供的函数在每个批处理间隔中被调用。这允许你做随时间变化的RDD操作，也就是说，RDD操作、分区数量、广播变量等可以在批处理之间改变。

Window Operations 窗口运算

Spark Streaming also provides windowed computations, which allow you to apply transformations over a sliding window of data. The following figure illustrates this sliding window.
Spark Streaming还提供了窗口计算，允许您在数据的滑动窗口上应用转换。下图说明了此滑动窗口。

As shown in the figure, every time the window slides over a source DStream, the source RDDs that fall within the window are combined and operated upon to produce the RDDs of the windowed DStream. In this specific case, the operation is applied over the last 3 time units of data, and slides by 2 time units. This shows that any window operation needs to specify two parameters.
如图所示，每次窗口在源DStream上滑动时，落在窗口内的源RDD被组合并操作以产生窗口化DStream的RDD。在这种特定情况下，该操作应用于数据的最后3个时间单位，并滑动2个时间单位。这表明任何窗口操作都需要指定两个参数。

• window length - The duration of the window (3 in the figure).
  窗口长度—窗口的持续时间（图中的3）。
• sliding interval - The interval at which the window operation is performed (2 in the figure).
  滑动间隔—执行窗口操作的间隔（图中的2）。

These two parameters must be multiples of the batch interval of the source DStream (1 in the figure).
这两个参数必须是源DStream的批处理间隔的倍数（图中的1）。

Let’s illustrate the window operations with an example. Say, you want to extend the earlier example by generating word counts over the last 30 seconds of data, every 10 seconds. To do this, we have to apply the reduceByKey operation on the pairs DStream of (word, 1) pairs over the last 30 seconds of data. This is done using the operation reduceByKeyAndWindow.
让我们用一个例子来说明窗口操作。比如，您想扩展前面的示例，每隔10秒生成最近30秒数据的单词计数。为此，我们必须在过去30秒的数据中对 (word, 1) 对的 pairs DStream应用 reduceByKey 操作。这是使用操作 reduceByKeyAndWindow 完成的。

• Scala
• Java
• Python

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

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

Some of the common window operations are as follows. All of these operations take the said two parameters - windowLength and slideInterval.
一些常见的窗口操作如下。所有这些操作都采用上述两个参数—windowLength和slideInterval。

Join Operations 连接操作

Finally, its worth highlighting how easily you can perform different kinds of joins in Spark Streaming.
最后，值得强调的是，您可以在Spark Streaming中轻松执行不同类型的连接。

Stream-stream joins 流—流连接

Streams can be very easily joined with other streams.
流可以很容易地与其他流连接。

• Scala
• Java
• Python

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 = ...
stream2 = ...
joinedStream = stream1.join(stream2)

Here, in each batch interval, the RDD generated by stream1 will be joined with the RDD generated by stream2. You can also do leftOuterJoin, rightOuterJoin, fullOuterJoin. Furthermore, it is often very useful to do joins over windows of the streams. That is pretty easy as well.
在这里，在每个批次间隔中，由 stream1 生成的RDD将与由 stream2 生成的RDD连接。你也可以做 leftOuterJoin 、 rightOuterJoin 、 fullOuterJoin 。此外，在流的窗口上进行连接通常非常有用。这也很容易。

• Scala
• Java
• Python

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

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

Stream-dataset joins 流数据集连接

This has already been shown earlier while explain DStream.transform operation. Here is yet another example of joining a windowed stream with a dataset.
这在之前解释 DStream.transform 操作时已经显示过。这是另一个将窗口化流与数据集连接的示例。

• Scala
• Java
• Python

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

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

In fact, you can also dynamically change the dataset you want to join against. The function provided to transform is evaluated every batch interval and therefore will use the current dataset that dataset reference points to.
事实上，您还可以动态更改要连接的数据集。提供给 transform 的函数在每个批处理间隔进行评估，因此将使用 dataset 引用指向的当前数据集。

The complete list of DStream transformations is available in the API documentation. For the Scala API, see DStream and PairDStreamFunctions. For the Java API, see JavaDStream and JavaPairDStream. For the Python API, see DStream.
完整的DStream转换列表可以在API文档中找到。有关Scala API，请参阅DStream和PairDStreamFunctions。有关Java API，请参见JavaDStream和JavaPairDStream。关于Python API，请参阅DStream。

Output Operations on DStreams
DStreams上的输出操作

Output operations allow DStream’s data to be pushed out to external systems like a database or a file systems. Since the output operations actually allow the transformed data to be consumed by external systems, they trigger the actual execution of all the DStream transformations (similar to actions for RDDs). Currently, the following output operations are defined:
输出操作允许DStream的数据被推送到外部系统，如数据库或文件系统。由于输出操作实际上允许转换后的数据被外部系统使用，因此它们触发了所有DStream转换的实际执行（类似于RDD的操作）。目前，定义了以下输出操作：

Design Patterns for using foreachRDD
使用foreachRDD的设计模式

dstream.foreachRDD is a powerful primitive that allows data to be sent out to external systems. However, it is important to understand how to use this primitive correctly and efficiently. Some of the common mistakes to avoid are as follows.
dstream.foreachRDD 是一个功能强大的原语，允许将数据发送到外部系统。但是，了解如何正确有效地使用此原语非常重要。以下是一些需要避免的常见错误。

Often writing data to external system requires creating a connection object (e.g. TCP connection to a remote server) and using it to send data to a remote system. For this purpose, a developer may inadvertently try creating a connection object at the Spark driver, and then try to use it in a Spark worker to save records in the RDDs. For example (in Scala),
通常，将数据写入外部系统需要创建连接对象（例如，到远程服务器的TCP连接）并使用它将数据发送到远程系统。为此，开发人员可能会无意中尝试在Spark驱动程序中创建一个连接对象，然后尝试在Spark worker中使用它来保存RDD中的记录。例如（在Scala中），

• Scala
• Java
• Python

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(rdd):
    connection = createNewConnection()  # executed at the driver
    rdd.foreach(lambda record: connection.send(record))
    connection.close()

dstream.foreachRDD(sendRecord)

This is incorrect as this requires the connection object to be serialized and sent from the driver to the worker. Such connection objects are rarely transferable across machines. This error may manifest as serialization errors (connection object not serializable), initialization errors (connection object needs to be initialized at the workers), etc. The correct solution is to create the connection object at the worker.
这是不正确的，因为这需要连接对象被序列化并从驱动程序发送到工作程序。这样的连接对象很少能跨机器传输。此错误可能表现为序列化错误（连接对象不可序列化）、初始化错误（连接对象需要在工作线程上初始化）等。正确的解决方案是在工作线程上创建连接对象。

However, this can lead to another common mistake - creating a new connection for every record. For example,
然而，这可能导致另一个常见的错误-为每个记录创建一个新连接。例如，

• Scala
• Java
• Python

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

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

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

Typically, creating a connection object has time and resource overheads. Therefore, creating and destroying a connection object for each record can incur unnecessarily high overheads and can significantly reduce the overall throughput of the system. A better solution is to use rdd.foreachPartition - create a single connection object and send all the records in a RDD partition using that connection.
通常，创建连接对象需要花费时间和资源。因此，为每个记录创建和销毁连接对象可能会导致不必要的高开销，并可能显著降低系统的总体吞吐量。更好的解决方案是使用 rdd.foreachPartition -创建一个单一的连接对象，并使用该连接发送RDD分区中的所有记录。

• Scala
• Java
• Python

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

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

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

This amortizes the connection creation overheads over many records.
这将连接创建开销分摊到许多记录上。

Finally, this can be further optimized by reusing connection objects across multiple RDDs/batches. One can maintain a static pool of connection objects than can be reused as RDDs of multiple batches are pushed to the external system, thus further reducing the overheads.
最后，这可以通过跨多个RDD/batch重用连接对象来进一步优化。人们可以维护一个静态的连接对象池，当多个批次的RDD被推送到外部系统时，这些连接对象可以被重用，从而进一步减少开销。

• Scala
• Java
• Python

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

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

Note that the connections in the pool should be lazily created on demand and timed out if not used for a while. This achieves the most efficient sending of data to external systems.
请注意，池中的连接应按需延迟创建，如果一段时间未使用，则会超时。这实现了向外部系统最有效地发送数据。

Other points to remember:
其他要记住的要点：

• DStreams are executed lazily by the output operations, just like RDDs are lazily executed by RDD actions. Specifically, RDD actions inside the DStream output operations force the processing of the received data. Hence, if your application does not have any output operation, or has output operations like dstream.foreachRDD() without any RDD action inside them, then nothing will get executed. The system will simply receive the data and discard it.
  DStream是由输出操作延迟执行的，就像RDD是由RDD操作延迟执行的一样。具体来说，DStream输出操作中的RDD操作强制处理接收到的数据。因此，如果你的应用程序没有任何输出操作，或者有像 dstream.foreachRDD() 这样的输出操作，而里面没有任何RDD操作，那么什么都不会被执行。系统将简单地接收数据并丢弃它。

• By default, output operations are executed one-at-a-time. And they are executed in the order they are defined in the application.
  默认情况下，输出操作一次执行一个。它们按照应用程序中定义的顺序执行。

DataFrame and SQL Operations
DataFrame和SQL操作

You can easily use DataFrames and SQL operations on streaming data. You have to create a SparkSession using the SparkContext that the StreamingContext is using. Furthermore, this has to done such that it can be restarted on driver failures. This is done by creating a lazily instantiated singleton instance of SparkSession. This is shown in the following example. It modifies the earlier word count example to generate word counts using DataFrames and SQL. Each RDD is converted to a DataFrame, registered as a temporary table and then queried using SQL.
您可以轻松地对流数据使用DataFrames和SQL操作。您必须使用StreamingContext正在使用的SparkContext创建SparkSession。此外，这必须这样做，以便它可以在驱动程序故障时重新启动。这是通过创建一个SparkSession的惰性实例化单例实例来完成的。这在下面的示例中显示。它修改了前面的单词计数示例，以使用DataFrames和SQL生成单词计数。每个RDD都被转换为DataFrame，注册为临时表，然后使用SQL进行查询。

• Scala
• Java
• Python

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

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

You can also run SQL queries on tables defined on streaming data from a different thread (that is, asynchronous to the running StreamingContext). Just make sure that you set the StreamingContext to remember a sufficient amount of streaming data such that the query can run. Otherwise the StreamingContext, which is unaware of the any asynchronous SQL queries, will delete off old streaming data before the query can complete. For example, if you want to query the last batch, but your query can take 5 minutes to run, then call streamingContext.remember(Minutes(5)) (in Scala, or equivalent in other languages).
您还可以对在不同线程（即与正在运行的StreamingContext异步）的流数据上定义的表运行SQL查询。只需确保您设置StreamingContext以记住足够数量的流数据，以便查询可以运行。否则，StreamingContext（它不知道任何异步SQL查询）将在查询完成之前删除旧的流数据。例如，如果你想查询最后一个批处理，但你的查询可能需要5分钟才能运行，那么调用 streamingContext.remember(Minutes(5)) （在Scala中，或在其他语言中等效）。

See the DataFrames and SQL guide to learn more about DataFrames.
请参阅DataFrames和SQL指南以了解有关DataFrames的更多信息。

MLlib Operations MLlib操作

You can also easily use machine learning algorithms provided by MLlib. First of all, there are streaming machine learning algorithms (e.g. Streaming Linear Regression, Streaming KMeans, etc.) which can simultaneously learn from the streaming data as well as apply the model on the streaming data. Beyond these, for a much larger class of machine learning algorithms, you can learn a learning model offline (i.e. using historical data) and then apply the model online on streaming data. See the MLlib guide for more details.
您还可以轻松使用MLlib提供的机器学习算法。首先，有流式机器学习算法（例如流式线性回归，流式KMeans等）。其可以同时从流数据学习以及将模型应用于流数据。除此之外，对于更大类别的机器学习算法，您可以离线学习学习模型（即使用历史数据），然后在线将模型应用于流数据。有关详细信息，请参阅MLlib指南。

Caching / Persistence 缓存/持久性

Similar to RDDs, DStreams also allow developers to persist the stream’s data in memory. That is, using the persist() method on a DStream will automatically persist every RDD of that DStream in memory. This is useful if the data in the DStream will be computed multiple times (e.g., multiple operations on the same data). For window-based operations like reduceByWindow and reduceByKeyAndWindow and state-based operations like updateStateByKey, this is implicitly true. Hence, DStreams generated by window-based operations are automatically persisted in memory, without the developer calling persist().
与RDD类似，DStreams也允许开发人员将流的数据持久化到内存中。也就是说，在DStream上使用 persist() 方法将自动将该DStream的每个RDD持久化到内存中。如果DStream中的数据将被多次计算（例如，对同一数据的多个操作）。对于像 reduceByWindow 和 reduceByKeyAndWindow 这样的基于窗口的操作以及像 updateStateByKey 这样的基于状态的操作，这是隐含的。因此，由基于窗口的操作生成的DStream会自动持久化在内存中，而无需开发人员调用 persist() 。

For input streams that receive data over the network (such as, Kafka, sockets, etc.), the default persistence level is set to replicate the data to two nodes for fault-tolerance.
对于通过网络接收数据的输入流（例如，Kafka、套接字等），默认持久性级别被设置为将数据复制到两个节点以用于容错。

Note that, unlike RDDs, the default persistence level of DStreams keeps the data serialized in memory. This is further discussed in the Performance Tuning section. More information on different persistence levels can be found in the Spark Programming Guide.
请注意，与RDD不同，DStreams的默认持久化级别将数据序列化在内存中。这将在"性能调优"一节中进一步讨论。关于不同持久性级别的更多信息可以在Spark编程指南中找到。

Checkpointing 检查点

A streaming application must operate 24/7 and hence must be resilient to failures unrelated to the application logic (e.g., system failures, JVM crashes, etc.). For this to be possible, Spark Streaming needs to checkpoint enough information to a fault- tolerant storage system such that it can recover from failures. There are two types of data that are checkpointed.
流应用必须24/7操作，因此必须对与应用逻辑无关的故障具有弹性（例如，系统故障、JVM崩溃等）。为了实现这一点，Spark Streaming需要将足够的信息检查点到容错存储系统，以便它可以从故障中恢复。有两种类型的数据被设置了检查点。

• Metadata checkpointing - Saving of the information defining the streaming computation to fault-tolerant storage like HDFS. This is used to recover from failure of the node running the driver of the streaming application (discussed in detail later). Metadata includes:
  元数据检查点-将定义流计算的信息保存到HDFS等容错存储中。这用于从运行流应用程序的驱动程序的节点的故障中恢复（稍后详细讨论）。元数据包括：
  • Configuration - The configuration that was used to create the streaming application.
    配置-用于创建流应用程序的配置。
  • DStream operations - The set of DStream operations that define the streaming application.
    DStream操作-定义流应用程序的DStream操作集。
  • Incomplete batches - Batches whose jobs are queued but have not completed yet.
    未完成批-作业已排队但尚未完成的批。
• Data checkpointing - Saving of the generated RDDs to reliable storage. This is necessary in some stateful transformations that combine data across multiple batches. In such transformations, the generated RDDs depend on RDDs of previous batches, which causes the length of the dependency chain to keep increasing with time. To avoid such unbounded increases in recovery time (proportional to dependency chain), intermediate RDDs of stateful transformations are periodically checkpointed to reliable storage (e.g. HDFS) to cut off the dependency chains.
  数据检查点-将生成的RDD保存到可靠的存储。这在某些跨多个批处理组合联合收割机数据的有状态转换中是必需的。在这样的转换中，生成的RDD依赖于先前批次的RDD，这导致依赖关系链的长度随时间不断增加。为了避免恢复时间的这种无限增加（与依赖性链成比例），有状态转换的中间RDD定期检查点到可靠存储（例如HDFS）以切断依赖性链。

To summarize, metadata checkpointing is primarily needed for recovery from driver failures, whereas data or RDD checkpointing is necessary even for basic functioning if stateful transformations are used.
总而言之，元数据检查点主要用于从驱动程序故障中恢复，而数据或RDD检查点即使对于使用有状态转换的基本功能也是必要的。

When to enable Checkpointing
何时启用检查点

Checkpointing must be enabled for applications with any of the following requirements:
必须为满足以下任何要求的应用程序启用检查点：

• Usage of stateful transformations - If either updateStateByKey or reduceByKeyAndWindow (with inverse function) is used in the application, then the checkpoint directory must be provided to allow for periodic RDD checkpointing.
  使用有状态转换-如果在应用程序中使用 updateStateByKey 或 reduceByKeyAndWindow （具有反函数），则必须提供检查点目录以允许定期RDD检查点。
• Recovering from failures of the driver running the application - Metadata checkpoints are used to recover with progress information.
  从运行应用程序的驱动程序的故障中恢复-元数据检查点用于使用进度信息进行恢复。

Note that simple streaming applications without the aforementioned stateful transformations can be run without enabling checkpointing. The recovery from driver failures will also be partial in that case (some received but unprocessed data may be lost). This is often acceptable and many run Spark Streaming applications in this way. Support for non-Hadoop environments is expected to improve in the future.
请注意，没有上述状态转换的简单流应用程序可以在不启用检查点的情况下运行。在这种情况下，从驱动程序故障中恢复也将是部分的（一些接收到但未处理的数据可能会丢失）。这通常是可以接受的，许多人以这种方式运行Spark Streaming应用程序。对非Hadoop环境的支持预计将在未来得到改善。

How to configure Checkpointing
如何配置检查点

Checkpointing can be enabled by setting a directory in a fault-tolerant, reliable file system (e.g., HDFS, S3, etc.) to which the checkpoint information will be saved. This is done by using streamingContext.checkpoint(checkpointDirectory). This will allow you to use the aforementioned stateful transformations. Additionally, if you want to make the application recover from driver failures, you should rewrite your streaming application to have the following behavior.
可以通过在容错、可靠的文件系统中设置目录来启用检查点（例如，HDFS、S3等）检查点信息将被保存到其中。这是通过使用 streamingContext.checkpoint(checkpointDirectory) 来实现的。这将允许您使用前面提到的有状态转换。另外，如果你想让应用程序从驱动程序故障中恢复，你应该重写你的流应用程序，使其具有以下行为。

• When the program is being started for the first time, it will create a new StreamingContext, set up all the streams and then call start().
  当程序第一次启动时，它将创建一个新的StreamingContext，设置所有流，然后调用start（）。
• When the program is being restarted after failure, it will re-create a StreamingContext from the checkpoint data in the checkpoint directory.
  当程序在失败后重新启动时，它将从检查点目录中的检查点数据重新创建StreamingContext。

• Scala
• Java
• Python

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

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

In addition to using getOrCreate one also needs to ensure that the driver process gets restarted automatically on failure. This can only be done by the deployment infrastructure that is used to run the application. This is further discussed in the Deployment section.
除了使用 getOrCreate 之外，还需要确保驱动程序进程在失败时自动重新启动。这只能由用于运行应用程序的部署基础结构来完成。这将在“部署”一节中进一步讨论。

Note that checkpointing of RDDs incurs the cost of saving to reliable storage. This may cause an increase in the processing time of those batches where RDDs get checkpointed. Hence, the interval of checkpointing needs to be set carefully. At small batch sizes (say 1 second), checkpointing every batch may significantly reduce operation throughput. Conversely, checkpointing too infrequently causes the lineage and task sizes to grow, which may have detrimental effects. For stateful transformations that require RDD checkpointing, the default interval is a multiple of the batch interval that is at least 10 seconds. It can be set by using dstream.checkpoint(checkpointInterval). Typically, a checkpoint interval of 5 - 10 sliding intervals of a DStream is a good setting to try.
请注意，RDD的检查点会导致节省到可靠存储的成本。这可能会导致RDD设置检查点的那些批次的处理时间增加。因此，检查点的间隔需要仔细设置。在小批量时（比如1秒），对每个批量设置检查点可能会显著降低操作吞吐量。相反，检查点太少会导致沿袭和任务大小增加，这可能会产生不利影响。对于需要RDD检查点的有状态转换，默认间隔是批处理间隔的倍数，至少为10秒。可以使用 dstream.checkpoint(checkpointInterval) 进行设置。通常，DStream的5 - 10个滑动间隔的检查点间隔是一个很好的尝试设置。

Accumulators, Broadcast Variables, and Checkpoints
累加器、广播变量和检查点

Accumulators and Broadcast variables cannot be recovered from checkpoint in Spark Streaming. If you enable checkpointing and use Accumulators or Broadcast variables as well, you’ll have to create lazily instantiated singleton instances for Accumulators and Broadcast variables so that they can be re-instantiated after the driver restarts on failure. This is shown in the following example.
累加器和广播变量无法从Spark Streaming中的检查点恢复。如果您启用检查点并同时使用Accumulators或Broadcast变量，则必须为Accumulators和Broadcast变量创建延迟实例化的单例实例，以便在驱动程序重启失败后重新实例化它们。这在下面的示例中显示。

• Scala
• Java
• Python

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

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)

Deploying Applications 部署应用程序

This section discusses the steps to deploy a Spark Streaming application.
本节讨论部署Spark Streaming应用程序的步骤。

Requirements 要求

To run a Spark Streaming applications, you need to have the following.
要运行Spark Streaming应用程序，您需要具备以下条件。

• Cluster with a cluster manager - This is the general requirement of any Spark application, and discussed in detail in the deployment guide.
  使用集群管理器集群—这是任何Spark应用程序的一般要求，在部署指南中有详细讨论。

• Package the application JAR - You have to compile your streaming application into a JAR. If you are using spark-submit to start the application, then you will not need to provide Spark and Spark Streaming in the JAR. However, if your application uses advanced sources (e.g. Kafka), then you will have to package the extra artifact they link to, along with their dependencies, in the JAR that is used to deploy the application. For example, an application using KafkaUtils will have to include spark-streaming-kafka-0-10_2.12 and all its transitive dependencies in the application JAR.
  将应用程序打包成一个包——你必须将你的流应用程序编译成一个包。如果您使用 spark-submit 来启动应用程序，那么您将不需要在SDK中提供Spark和Spark Streaming。但是，如果您的应用程序使用高级源（例如Kafka），那么您必须将它们链接到的额外工件沿着它们的依赖项打包到用于部署应用程序的目录中。例如，使用 KafkaUtils 的应用程序必须在应用程序中包含 spark-streaming-kafka-0-10_2.12 及其所有可传递依赖项。

• Configuring sufficient memory for the executors - Since the received data must be stored in memory, the executors must be configured with sufficient memory to hold the received data. Note that if you are doing 10 minute window operations, the system has to keep at least last 10 minutes of data in memory. So the memory requirements for the application depends on the operations used in it.
  为执行器配置足够的内存-由于接收的数据必须存储在内存中，因此执行器必须配置足够的内存来保存接收的数据。请注意，如果您正在执行10分钟的窗口操作，则系统必须在内存中保留至少最后10分钟的数据。因此，应用程序的内存需求取决于其中使用的操作。

• Configuring checkpointing - If the stream application requires it, then a directory in the Hadoop API compatible fault-tolerant storage (e.g. HDFS, S3, etc.) must be configured as the checkpoint directory and the streaming application written in a way that checkpoint information can be used for failure recovery. See the checkpointing section for more details.
  配置检查点-如果流应用程序需要，则在Hadoop API兼容的容错存储（例如HDFS、S3等）中的目录必须配置为检查点目录，并以检查点信息可用于故障恢复的方式编写流应用程序。有关更多详细信息，请参见检查点部分。

• Configuring automatic restart of the application driver - To automatically recover from a driver failure, the deployment infrastructure that is used to run the streaming application must monitor the driver process and relaunch the driver if it fails. Different cluster managers have different tools to achieve this.
  配置应用程序驱动程序的自动重新启动-要从驱动程序故障中自动恢复，用于运行流式应用程序的部署基础结构必须监视驱动程序进程，并在驱动程序失败时重新启动驱动程序。不同的集群管理器有不同的工具来实现这一点。
  • Spark Standalone - A Spark application driver can be submitted to run within the Spark Standalone cluster (see cluster deploy mode), that is, the application driver itself runs on one of the worker nodes. Furthermore, the Standalone cluster manager can be instructed to supervise the driver, and relaunch it if the driver fails either due to non-zero exit code, or due to failure of the node running the driver. See cluster mode and supervise in the Spark Standalone guide for more details.
    Spark Standalone -可以提交Spark应用程序驱动程序以在Spark Standalone集群中运行（参见集群部署模式），也就是说，应用程序驱动程序本身在其中一个工作节点上运行。此外，可以指示独立集群管理器来监督驱动程序，并且如果驱动程序由于非零退出代码或由于运行驱动程序的节点的故障而失败，则重新启动驱动程序。有关详细信息，请参阅Spark Standalone指南中的集群模式和监督。
  • YARN - Yarn supports a similar mechanism for automatically restarting an application. Please refer to YARN documentation for more details.
    YARN - Yarn支持类似的机制来自动重启应用程序。请参阅YARN文档了解更多详细信息。
  • Mesos - Marathon has been used to achieve this with Mesos.
    Mesos -马拉松已被用于通过Mesos实现这一点。

• Configuring write-ahead logs - Since Spark 1.2, we have introduced write-ahead logs for achieving strong fault-tolerance guarantees. If enabled, all the data received from a receiver gets written into a write-ahead log in the configuration checkpoint directory. This prevents data loss on driver recovery, thus ensuring zero data loss (discussed in detail in the Fault-tolerance Semantics section). This can be enabled by setting the configuration parameter spark.streaming.receiver.writeAheadLog.enable to true. However, these stronger semantics may come at the cost of the receiving throughput of individual receivers. This can be corrected by running more receivers in parallel to increase aggregate throughput. Additionally, it is recommended that the replication of the received data within Spark be disabled when the write-ahead log is enabled as the log is already stored in a replicated storage system. This can be done by setting the storage level for the input stream to StorageLevel.MEMORY_AND_DISK_SER. While using S3 (or any file system that does not support flushing) for write-ahead logs, please remember to enable spark.streaming.driver.writeAheadLog.closeFileAfterWrite and spark.streaming.receiver.writeAheadLog.closeFileAfterWrite. See Spark Streaming Configuration for more details. Note that Spark will not encrypt data written to the write-ahead log when I/O encryption is enabled. If encryption of the write-ahead log data is desired, it should be stored in a file system that supports encryption natively.
  配置写前日志-从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 Configuration了解更多详情。请注意，当启用I/O加密时，Spark不会加密写入到写前日志的数据。如果需要对预写日志数据进行加密，则应将其存储在本机支持加密的文件系统中。

• Setting the max receiving rate - If the cluster resources is not large enough for the streaming application to process data as fast as it is being received, the receivers can be rate limited by setting a maximum rate limit in terms of records / sec. See the configuration parameters spark.streaming.receiver.maxRate for receivers and spark.streaming.kafka.maxRatePerPartition for Direct Kafka approach. In Spark 1.5, we have introduced a feature called backpressure that eliminate the need to set this rate limit, as Spark Streaming automatically figures out the rate limits and dynamically adjusts them if the processing conditions change. This backpressure can be enabled by setting the configuration parameter spark.streaming.backpressure.enabled to true.
  设置最大接收速率-如果群集资源不足以让流应用程序以接收数据的速度处理数据，则可以通过设置记录/秒的最大速率限制来限制接收器的速率。请参阅用于接收器的配置参数 spark.streaming.receiver.maxRate 和用于直接Kafka方法的配置参数 spark.streaming.kafka.maxRatePerPartition 。在Spark 1.5中，我们引入了一个称为背压的功能，它消除了设置此速率限制的需要，因为Spark Streaming会自动计算速率限制，并在处理条件发生变化时动态调整它们。可通过将配置参数 spark.streaming.backpressure.enabled 设置为 true 来启用此背压。

Upgrading Application Code
升级应用程序代码

If a running Spark Streaming application needs to be upgraded with new application code, then there are two possible mechanisms.
如果一个正在运行的Spark Streaming应用程序需要使用新的应用程序代码进行升级，那么有两种可能的机制。

• The upgraded Spark Streaming application is started and run in parallel to the existing application. Once the new one (receiving the same data as the old one) has been warmed up and is ready for prime time, the old one be can be brought down. Note that this can be done for data sources that support sending the data to two destinations (i.e., the earlier and upgraded applications).
  升级后的Spark Streaming应用程序将与现有应用程序并行启动和运行。一旦新的一个（接收与旧的相同的数据）已经预热并准备好了黄金时间，旧的一个可以被关闭。注意，这可以对支持将数据发送到两个目的地（即，早期和升级的应用程序）。

• The existing application is shutdown gracefully (see StreamingContext.stop(...) or JavaStreamingContext.stop(...) for graceful shutdown options) which ensure data that has been received is completely processed before shutdown. Then the upgraded application can be started, which will start processing from the same point where the earlier application left off. Note that this can be done only with input sources that support source-side buffering (like Kafka) as data needs to be buffered while the previous application was down and the upgraded application is not yet up. And restarting from earlier checkpoint information of pre-upgrade code cannot be done. The checkpoint information essentially contains serialized Scala/Java/Python objects and trying to deserialize objects with new, modified classes may lead to errors. In this case, either start the upgraded app with a different checkpoint directory, or delete the previous checkpoint directory.
  现有应用程序正常关闭（有关正常关闭选项，请参阅 StreamingContext.stop(...) 或 JavaStreamingContext.stop(...) ），这确保在关闭之前完全处理已接收的数据。然后可以启动升级后的应用程序，这将从先前应用程序停止的同一点开始处理。请注意，这只能使用支持源端缓冲的输入源（如Kafka）来完成，因为在先前的应用程序关闭而升级的应用程序尚未启动时，需要缓冲数据。并且无法从升级前代码的早期检查点信息重新启动。检查点信息基本上包含序列化的Scala/Java/Python对象，尝试使用新的修改后的类来对对象进行序列化可能会导致错误。在这种情况下，请使用其他检查点目录启动升级后的应用，或删除以前的检查点目录。

Monitoring Applications 监测应用

Beyond Spark’s monitoring capabilities, there are additional capabilities specific to Spark Streaming. When a StreamingContext is used, the Spark web UI shows an additional Streaming tab which shows statistics about running receivers (whether receivers are active, number of records received, receiver error, etc.) and completed batches (batch processing times, queueing delays, etc.). This can be used to monitor the progress of the streaming application.
除了Spark的监控功能之外，还有Spark Streaming特有的其他功能。当使用StreamingContext时，Spark Web UI会显示一个额外的 Streaming 选项卡，该选项卡显示有关正在运行的接收器的统计信息（接收器是否处于活动状态，接收到的记录数，接收器错误等）。和已完成的批次（批次处理时间、装配延迟等）。这可以用于监控流应用程序的进度。

The following two metrics in web UI are particularly important:
Web UI中的以下两个指标尤为重要：

• Processing Time - The time to process each batch of data.
  处理时间—处理每批数据的时间。
• Scheduling Delay - the time a batch waits in a queue for the processing of previous batches to finish.
  调度延迟—批处理在队列中等待前一批处理完成的时间。

If the batch processing time is consistently more than the batch interval and/or the queueing delay keeps increasing, then it indicates that the system is not able to process the batches as fast they are being generated and is falling behind. In that case, consider reducing the batch processing time.
如果批处理时间始终大于批处理间隔和/或批处理延迟持续增加，则表明系统无法像生成批处理那样快速地处理批处理，并且落后了。在这种情况下，考虑减少批处理时间。

The progress of a Spark Streaming program can also be monitored using the StreamingListener interface, which allows you to get receiver status and processing times. Note that this is a developer API and it is likely to be improved upon (i.e., more information reported) in the future.
Spark Streaming程序的进度也可以使用StreamingStream界面进行监控，该界面允许您获取接收器状态和处理时间。请注意，这是一个开发人员API，并且可能会得到改进（即，更多信息在未来报道。

Performance Tuning 性能调优

Getting the best performance out of a Spark Streaming application on a cluster requires a bit of tuning. This section explains a number of the parameters and configurations that can be tuned to improve the performance of you application. At a high level, you need to consider two things:
在集群上获得Spark Streaming应用程序的最佳性能需要一些调优。本节解释了一些参数和配置，这些参数和配置可以通过调整来提高应用程序的性能。在高层次上，你需要考虑两件事：

1. Reducing the processing time of each batch of data by efficiently using cluster resources.
   通过有效地使用群集资源减少每批数据的处理时间。

2. Setting the right batch size such that the batches of data can be processed as fast as they are received (that is, data processing keeps up with the data ingestion).
   设置正确的批量大小，以便批量数据可以像接收到它们一样快地处理（即，数据处理与数据摄取保持同步）。

Reducing the Batch Processing Times
减少批处理时间

There are a number of optimizations that can be done in Spark to minimize the processing time of each batch. These have been discussed in detail in the Tuning Guide. This section highlights some of the most important ones.
在Spark中可以进行许多优化，以最大限度地减少每个批次的处理时间。这些在《调音指南》中有详细讨论。本节重点介绍了其中一些最重要的问题。

Level of Parallelism in Data Receiving
数据接收的公平性水平

Receiving data over the network (like Kafka, socket, etc.) requires the data to be deserialized and stored in Spark. If the data receiving becomes a bottleneck in the system, then consider parallelizing the data receiving. Note that each input DStream creates a single receiver (running on a worker machine) that receives a single stream of data. Receiving multiple data streams can therefore be achieved by creating multiple input DStreams and configuring them to receive different partitions of the data stream from the source(s). For example, a single Kafka input DStream receiving two topics of data can be split into two Kafka input streams, each receiving only one topic. This would run two receivers, allowing data to be received in parallel, thus increasing overall throughput. These multiple DStreams can be unioned together to create a single DStream. Then the transformations that were being applied on a single input DStream can be applied on the unified stream. This is done as follows.
通过网络接收数据（如Kafka、socket等）需要将数据存储在Spark中。如果数据接收成为系统中的瓶颈，那么可以考虑将数据接收并行化。请注意，每个输入DStream创建一个接收器（运行在工作机器上），用于接收单个数据流。因此，可以通过创建多个输入DStream并将其配置为从源接收数据流的不同分区来实现接收多个数据流。例如，接收两个数据主题的单个Kafka输入DStream可以被分成两个Kafka输入流，每个输入流仅接收一个主题。这将运行两个接收器，允许并行接收数据，从而增加整体吞吐量。这些多个DStream可以联合在一起以创建单个DStream。然后，应用于单个输入DStream的转换可以应用于统一流。这是按照如下方式完成的。

• Scala
• Java
• Python

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

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

Another parameter that should be considered is the receiver’s block interval, which is determined by the configuration parameter spark.streaming.blockInterval. For most receivers, the received data is coalesced together into blocks of data before storing inside Spark’s memory. The number of blocks in each batch determines the number of tasks that will be used to process the received data in a map-like transformation. The number of tasks per receiver per batch will be approximately (batch interval / block interval). For example, block interval of 200 ms will create 10 tasks per 2 second batches. If the number of tasks is too low (that is, less than the number of cores per machine), then it will be inefficient as all available cores will not be used to process the data. To increase the number of tasks for a given batch interval, reduce the block interval. However, the recommended minimum value of block interval is about 50 ms, below which the task launching overheads may be a problem.
应当考虑的另一个参数是接收器的块间隔，其由配置参数 spark.streaming.blockInterval 确定。对于大多数接收器来说，接收到的数据在存储到Spark的内存之前会合并成数据块。每个批处理中的块数决定了在类似于映射的转换中处理接收到的数据所使用的任务数。每批每个接收器的任务数大约为（批间隔/块间隔）。例如，200 ms的块间隔将每2秒创建10个任务批。如果任务的数量太少（即，少于每台机器的核心数量），那么它将是低效的，因为所有可用的核心将不会被用于处理数据。若要增加给定批处理间隔的任务数，请减小块间隔。然而，块间隔的推荐最小值是大约50 ms，低于该值，任务启动开销可能是一个问题。

An alternative to receiving data with multiple input streams / receivers is to explicitly repartition the input data stream (using inputStream.repartition(<number of partitions>)). This distributes the received batches of data across the specified number of machines in the cluster before further processing.
用多个输入流/接收器接收数据的替代方案是显式地重新划分输入数据流（使用 inputStream.repartition(<number of partitions>) ）。这将在进一步处理之前将接收到的数据批分布到集群中指定数量的机器上。

For direct stream, please refer to Spark Streaming + Kafka Integration Guide
直播请参考Spark Streaming + Kafka集成指南

Level of Parallelism in Data Processing
数据处理中的层次分析法

Cluster resources can be under-utilized if the number of parallel tasks used in any stage of the computation is not high enough. For example, for distributed reduce operations like reduceByKey and reduceByKeyAndWindow, the default number of parallel tasks is controlled by the spark.default.parallelism configuration property. You can pass the level of parallelism as an argument (see PairDStreamFunctions documentation), or set the spark.default.parallelism configuration property to change the default.
如果在计算的任何阶段中使用的并行任务的数量不够高，则集群资源可能未得到充分利用。例如，对于像 reduceByKey 和 reduceByKeyAndWindow 这样的分布式reduce操作，并行任务的默认数量由 spark.default.parallelism 配置属性控制。您可以将并行度级别作为参数传递（参见 PairDStreamFunctions 文档），或者设置 spark.default.parallelism 配置属性以更改默认值。

Data Serialization 数据序列化

The overheads of data serialization can be reduced by tuning the serialization formats. In the case of streaming, there are two types of data that are being serialized.
通过调优序列化格式可以减少数据序列化的开销。在流式传输的情况下，有两种类型的数据正在被序列化。

• Input data: By default, the input data received through Receivers is stored in the executors’ memory with StorageLevel.MEMORY_AND_DISK_SER_2. That is, the data is serialized into bytes to reduce GC overheads, and replicated for tolerating executor failures. Also, the data is kept first in memory, and spilled over to disk only if the memory is insufficient to hold all of the input data necessary for the streaming computation. This serialization obviously has overheads – the receiver must deserialize the received data and re-serialize it using Spark’s serialization format.
  输入数据：默认情况下，通过Receivers接收的输入数据存储在执行器的内存中，并带有EQUIPLevel.MEMORY_AND_DISK_SER_2。也就是说，数据被序列化为字节以减少GC开销，并被复制以容忍执行器故障。此外，数据首先保存在内存中，只有当内存不足以容纳流式计算所需的所有输入数据时才溢出到磁盘。这种序列化显然有开销—接收方必须将接收到的数据重新序列化，并使用Spark的序列化格式重新序列化。

• Persisted RDDs generated by Streaming Operations: RDDs generated by streaming computations may be persisted in memory. For example, window operations persist data in memory as they would be processed multiple times. However, unlike the Spark Core default of StorageLevel.MEMORY_ONLY, persisted RDDs generated by streaming computations are persisted with StorageLevel.MEMORY_ONLY_SER (i.e. serialized) by default to minimize GC overheads.
  由流式传输操作生成的持久化RDD：由流式传输计算生成的RDD可以持久化在存储器中。例如，窗口操作将数据保存在内存中，因为它们将被多次处理。但是，与Spark Core默认的.MEMORY_ONLY不同，流计算生成的持久化RDD默认使用.MEMORY_ONLY_SER持久化（即序列化），以最大限度地减少GC开销。

In both cases, using Kryo serialization can reduce both CPU and memory overheads. See the Spark Tuning Guide for more details. For Kryo, consider registering custom classes, and disabling object reference tracking (see Kryo-related configurations in the Configuration Guide).
在这两种情况下，使用Kryo序列化可以减少CPU和内存开销。请参阅Spark Tuning Guide了解更多详情。对于Kryo，请考虑注册自定义类，并禁用对象引用跟踪（请参阅《配置指南》中与Kryo相关的配置）。

In specific cases where the amount of data that needs to be retained for the streaming application is not large, it may be feasible to persist data (both types) as deserialized objects without incurring excessive GC overheads. For example, if you are using batch intervals of a few seconds and no window operations, then you can try disabling serialization in persisted data by explicitly setting the storage level accordingly. This would reduce the CPU overheads due to serialization, potentially improving performance without too much GC overheads.
在需要为流应用程序保留的数据量不大的特定情况下，将数据（两种类型）持久化为实体化对象而不会产生过多的GC开销可能是可行的。例如，如果您使用的批处理间隔为几秒，并且没有窗口操作，则可以尝试通过显式设置相应的存储级别来禁用持久化数据中的序列化。这将减少由于序列化而产生的CPU开销，从而可能在不增加太多GC开销的情况下提高性能。

Task Launching Overheads
任务启动开销

If the number of tasks launched per second is high (say, 50 or more per second), then the overhead of sending out tasks to the executors may be significant and will make it hard to achieve sub-second latencies. The overhead can be reduced by the following changes:
如果每秒启动的任务数量很高（比如每秒50个或更多），那么向执行器发送任务的开销可能会很大，并且很难实现亚秒延迟。通过以下更改可以减少开销：

• Execution mode: Running Spark in Standalone mode or coarse-grained Mesos mode leads to better task launch times than the fine-grained Mesos mode. Please refer to the Running on Mesos guide for more details.
  执行模式：在独立模式或粗粒度Mesos模式下运行Spark会比细粒度Mesos模式更好地启动任务。请参阅在Mesos上运行指南了解更多详情。

These changes may reduce batch processing time by 100s of milliseconds, thus allowing sub-second batch size to be viable.
这些变化可以将批处理时间减少数百毫秒，从而使亚秒级的批处理大小变得可行。

Setting the Right Batch Interval
设置正确的批处理间隔

For a Spark Streaming application running on a cluster to be stable, the system should be able to process data as fast as it is being received. In other words, batches of data should be processed as fast as they are being generated. Whether this is true for an application can be found by monitoring the processing times in the streaming web UI, where the batch processing time should be less than the batch interval.
对于在集群上运行的稳定的Spark Streaming应用程序，系统应该能够像接收数据一样快地处理数据。换句话说，批量数据的处理速度应该和它们生成的速度一样快。对于应用程序来说，这是否成立可以通过监视流式Web UI中的处理时间来确定，其中批处理时间应该小于批处理间隔。

Depending on the nature of the streaming computation, the batch interval used may have significant impact on the data rates that can be sustained by the application on a fixed set of cluster resources. For example, let us consider the earlier WordCountNetwork example. For a particular data rate, the system may be able to keep up with reporting word counts every 2 seconds (i.e., batch interval of 2 seconds), but not every 500 milliseconds. So the batch interval needs to be set such that the expected data rate in production can be sustained.
根据流式计算的性质，使用的批处理间隔可能会对应用程序在固定的群集资源集上可以维持的数据速率产生重大影响。例如，让我们考虑前面的WordCountNetwork示例。对于特定的数据速率，系统可能能够跟上每2秒报告字计数（即，批处理间隔为2秒），但不是每500毫秒。因此，需要设置批处理间隔，以便可以维持生产中的预期数据速率。

A good approach to figure out the right batch size for your application is to test it with a conservative batch interval (say, 5-10 seconds) and a low data rate. To verify whether the system is able to keep up with the data rate, you can check the value of the end-to-end delay experienced by each processed batch (either look for “Total delay” in Spark driver log4j logs, or use the StreamingListener interface). If the delay is maintained to be comparable to the batch size, then system is stable. Otherwise, if the delay is continuously increasing, it means that the system is unable to keep up and it therefore unstable. Once you have an idea of a stable configuration, you can try increasing the data rate and/or reducing the batch size. Note that a momentary increase in the delay due to temporary data rate increases may be fine as long as the delay reduces back to a low value (i.e., less than batch size).
为您的应用程序确定正确的批处理大小的一个好方法是使用保守的批处理间隔（例如，5-10秒）和低数据速率对其进行测试。为了验证系统是否能够跟上数据速率，您可以检查每个处理的批处理所经历的端到端延迟的值（在Spark驱动程序log4j日志中查找“Total delay”，或者使用StreamingData接口）。如果延迟保持与批量相当，则系统稳定。否则，如果延迟持续增加，这意味着系统无法跟上，因此不稳定。一旦你有了一个稳定的配置的想法，你可以尝试增加数据速率和/或减少批量大小。注意，由于临时数据速率增加而引起的延迟的瞬时增加可以是好的，只要延迟减小回到低值（即，小于批量）。

Memory Tuning 内存调优

Tuning the memory usage and GC behavior of Spark applications has been discussed in great detail in the Tuning Guide. It is strongly recommended that you read that. In this section, we discuss a few tuning parameters specifically in the context of Spark Streaming applications.
调优指南中详细讨论了Spark应用程序的内存使用和GC行为。强烈建议你读一下。在本节中，我们将讨论一些专门针对Spark Streaming应用程序的调优参数。

The amount of cluster memory required by a Spark Streaming application depends heavily on the type of transformations used. For example, if you want to use a window operation on the last 10 minutes of data, then your cluster should have sufficient memory to hold 10 minutes worth of data in memory. Or if you want to use updateStateByKey with a large number of keys, then the necessary memory will be high. On the contrary, if you want to do a simple map-filter-store operation, then the necessary memory will be low.
Spark Streaming应用程序所需的集群内存量在很大程度上取决于所使用的转换类型。例如，如果您想对最近10分钟的数据使用窗口操作，那么您的集群应该有足够的内存来保存10分钟的数据。或者，如果你想用大量的键来使用 updateStateByKey ，那么所需的内存就会很高。相反，如果你想做一个简单的映射-过滤-存储操作，那么所需的内存将很低。

In general, since the data received through receivers is stored with StorageLevel.MEMORY_AND_DISK_SER_2, the data that does not fit in memory will spill over to the disk. This may reduce the performance of the streaming application, and hence it is advised to provide sufficient memory as required by your streaming application. Its best to try and see the memory usage on a small scale and estimate accordingly.
一般来说，由于通过接收器接收的数据存储在磁盘级别. MEMORY_AND_DISK_SER_2中，因此内存中不适合的数据将溢出到磁盘。这可能会降低流媒体应用程序的性能，因此建议根据流媒体应用程序的要求提供足够的内存。最好尝试在小范围内查看内存使用情况并进行相应的估计。

Another aspect of memory tuning is garbage collection. For a streaming application that requires low latency, it is undesirable to have large pauses caused by JVM Garbage Collection.
内存调优的另一个方面是垃圾收集。对于需要低延迟的流应用程序，不希望JVM垃圾收集导致大量暂停。

There are a few parameters that can help you tune the memory usage and GC overheads:
有几个参数可以帮助您优化内存使用和GC开销：

• Persistence Level of DStreams: As mentioned earlier in the Data Serialization section, the input data and RDDs are by default persisted as serialized bytes. This reduces both the memory usage and GC overheads, compared to deserialized persistence. Enabling Kryo serialization further reduces serialized sizes and memory usage. Further reduction in memory usage can be achieved with compression (see the Spark configuration spark.rdd.compress), at the cost of CPU time.
  DStreams的持久化级别：正如前面在数据序列化一节中所提到的，输入数据和RDD默认情况下作为序列化字节持久化。与并行持久化相比，这减少了内存使用和GC开销。启用Kryo序列化进一步减少了序列化的大小和内存使用。通过压缩可以进一步减少内存使用（参见Spark配置 spark.rdd.compress ），但代价是CPU时间。

• Clearing old data: By default, all input data and persisted RDDs generated by DStream transformations are automatically cleared. Spark Streaming decides when to clear the data based on the transformations that are used. For example, if you are using a window operation of 10 minutes, then Spark Streaming will keep around the last 10 minutes of data, and actively throw away older data. Data can be retained for a longer duration (e.g. interactively querying older data) by setting streamingContext.remember.
  清除旧数据：默认情况下，所有输入数据和DStream转换生成的持久化RDD都会自动清除。Spark Streaming根据所使用的转换决定何时清除数据。例如，如果您使用的是10分钟的窗口操作，那么Spark Streaming将保留最后10分钟的数据，并主动丢弃旧数据。通过设置 streamingContext.remember ，数据可以保留更长的时间（例如交互式查询旧数据）。

• CMS Garbage Collector: Use of the concurrent mark-and-sweep GC is strongly recommended for keeping GC-related pauses consistently low. Even though concurrent GC is known to reduce the overall processing throughput of the system, its use is still recommended to achieve more consistent batch processing times. Make sure you set the CMS GC on both the driver (using --driver-java-options in spark-submit) and the executors (using Spark configuration spark.executor.extraJavaOptions).
  CMS垃圾收集器：强烈建议使用并发标记和清除GC，以保持GC相关的暂停始终较低。尽管已知并发GC会降低系统的总体处理吞吐量，但仍建议使用它来实现更一致的批处理时间。确保在驱动程序（使用 spark-submit 中的 --driver-java-options ）和执行器（使用Spark配置 spark.executor.extraJavaOptions ）上都设置了CMS GC。

• Other tips: To further reduce GC overheads, here are some more tips to try.
  其他提示：为了进一步减少GC开销，这里还有一些可以尝试的提示。

  • Persist RDDs using the OFF_HEAP storage level. See more detail in the Spark Programming Guide.
    使用 OFF_HEAP 存储级别持久化RDD。更多细节请参见Spark编程指南。
  • Use more executors with smaller heap sizes. This will reduce the GC pressure within each JVM heap.
    使用更多的具有较小堆大小的执行器。这将减少每个JVM堆中的GC压力。

Important points to remember:
要记住的要点：

• A DStream is associated with a single receiver. For attaining read parallelism multiple receivers i.e. multiple DStreams need to be created. A receiver is run within an executor. It occupies one core. Ensure that there are enough cores for processing after receiver slots are booked i.e. spark.cores.max should take the receiver slots into account. The receivers are allocated to executors in a round robin fashion.
  DStream与单个接收器相关联。为了获得读取并行性，需要创建多个接收器，即多个DStream。接收器在执行器中运行。它占据了一个核心。确保在预订接收器插槽后有足够的内核用于处理，即 spark.cores.max 应考虑接收器插槽。接收器以循环方式分配给执行器。

• When data is received from a stream source, receiver creates blocks of data. A new block of data is generated every blockInterval milliseconds. N blocks of data are created during the batchInterval where N = batchInterval/blockInterval. These blocks are distributed by the BlockManager of the current executor to the block managers of other executors. After that, the Network Input Tracker running on the driver is informed about the block locations for further processing.
  当从流源接收数据时，接收器创建数据块。每隔blockInterval毫秒生成一个新的数据块。在batchInterval期间创建N个数据块，其中N = batchInterval/blockInterval。这些块由当前执行器的BlockManager分发给其他执行器的块管理器。之后，在驱动程序上运行的网络输入跟踪器被告知块的位置，以便进一步处理。

• An RDD is created on the driver for the blocks created during the batchInterval. The blocks generated during the batchInterval are partitions of the RDD. Each partition is a task in spark. blockInterval== batchinterval would mean that a single partition is created and probably it is processed locally.
  在驱动程序上为batchInterval期间创建的块创建RDD。在batchInterval期间生成的块是RDD的分区。每个分区都是Spark中的一个任务。blockInterval== batchinterval意味着创建了一个单独的分区，并且可能是在本地处理的。

• The map tasks on the blocks are processed in the executors (one that received the block, and another where the block was replicated) that has the blocks irrespective of block interval, unless non-local scheduling kicks in. Having bigger blockinterval means bigger blocks. A high value of spark.locality.wait increases the chance of processing a block on the local node. A balance needs to be found out between these two parameters to ensure that the bigger blocks are processed locally.
  块上的映射任务在执行器（一个接收块，另一个复制块）中处理，该执行器具有块而不管块间隔，除非非本地调度启动。拥有更大的块间隔意味着更大的块。高值 spark.locality.wait 增加了在本地节点上处理块的机会。需要在这两个参数之间找到平衡，以确保在本地处理较大的块。

• Instead of relying on batchInterval and blockInterval, you can define the number of partitions by calling inputDstream.repartition(n). This reshuffles the data in RDD randomly to create n number of partitions. Yes, for greater parallelism. Though comes at the cost of a shuffle. An RDD’s processing is scheduled by driver’s jobscheduler as a job. At a given point of time only one job is active. So, if one job is executing the other jobs are queued.
  不依赖于batchInterval和blockInterval，您可以通过调用 inputDstream.repartition(n) 来定义分区的数量。这将随机地重新排列RDD中的数据以创建n个分区。是的，为了更好的并行性。不过，这是以洗牌为代价的。RDD的处理由驱动程序的jobscheduler作为作业调度。在给定的时间点，只有一个作业处于活动状态。因此，如果一个作业正在执行，则其他作业将排队。

• If you have two dstreams there will be two RDDs formed and there will be two jobs created which will be scheduled one after the another. To avoid this, you can union two dstreams. This will ensure that a single unionRDD is formed for the two RDDs of the dstreams. This unionRDD is then considered as a single job. However, the partitioning of the RDDs is not impacted.
  如果您有两个数据流，则将形成两个RDD，并且将创建两个作业，这两个作业将一个接一个地进行调度。为了避免这种情况，你可以合并两个数据流。这将确保为数据流的两个RDD形成单个unionRDD。然后将该unionRDD视为单个作业。但是，RDD的分区不受影响。

• If the batch processing time is more than batchinterval then obviously the receiver’s memory will start filling up and will end up in throwing exceptions (most probably BlockNotFoundException). Currently, there is no way to pause the receiver. Using SparkConf configuration spark.streaming.receiver.maxRate, rate of receiver can be limited.
  如果批处理时间超过batchinterval，那么很明显接收器的内存将开始填满，并最终抛出异常（最有可能是BlockNotFoundException）。目前，没有办法暂停接收器。使用SparkConf配置 spark.streaming.receiver.maxRate ，可以限制接收器的速率。

Fault-tolerance Semantics
容错语义

In this section, we will discuss the behavior of Spark Streaming applications in the event of failures.
在本节中，我们将讨论Spark Streaming应用程序在发生故障时的行为。

Background 背景

To understand the semantics provided by Spark Streaming, let us remember the basic fault-tolerance semantics of Spark’s RDDs.
为了理解Spark Streaming提供的语义，让我们记住Spark RDD的基本容错语义。

1. An RDD is an immutable, deterministically re-computable, distributed dataset. Each RDD remembers the lineage of deterministic operations that were used on a fault-tolerant input dataset to create it.
   RDD是一个不可变的，确定性可重新计算的分布式数据集。每个RDD都记住了在容错输入数据集上用于创建它的确定性操作的谱系。
2. If any partition of an RDD is lost due to a worker node failure, then that partition can be re-computed from the original fault-tolerant dataset using the lineage of operations.
   如果RDD的任何分区由于工作节点故障而丢失，则可以使用操作的谱系从原始容错数据集重新计算该分区。
3. Assuming that all of the RDD transformations are deterministic, the data in the final transformed RDD will always be the same irrespective of failures in the Spark cluster.
   假设所有的RDD转换都是确定性的，那么最终转换的RDD中的数据将始终相同，而不管Spark集群中的故障如何。

Spark operates on data in fault-tolerant file systems like HDFS or S3. Hence, all of the RDDs generated from the fault-tolerant data are also fault-tolerant. However, this is not the case for Spark Streaming as the data in most cases is received over the network (except when fileStream is used). To achieve the same fault-tolerance properties for all of the generated RDDs, the received data is replicated among multiple Spark executors in worker nodes in the cluster (default replication factor is 2). This leads to two kinds of data in the system that need to recovered in the event of failures:
Spark在HDFS或S3等容错文件系统中操作数据。因此，从容错数据生成的所有RDD也是容错的。然而，Spark Streaming的情况并非如此，因为大多数情况下数据都是通过网络接收的（使用 fileStream 时除外）。为了使所有生成的RDD都具有相同的容错属性，接收到的数据会在集群中工作节点的多个Spark执行器之间进行复制（默认复制因子为2）。这导致系统中有两种数据需要在发生故障时恢复：

1. Data received and replicated - This data survives failure of a single worker node as a copy of it exists on one of the other nodes.
   接收和复制的数据-此数据在单个工作节点发生故障时仍然存在，因为它的副本存在于其他节点上。
2. Data received but buffered for replication - Since this is not replicated, the only way to recover this data is to get it again from the source.
   已接收但已缓冲以供复制的数据-由于这是不复制的，因此恢复此数据的唯一方法是再次从源获取它。

Furthermore, there are two kinds of failures that we should be concerned about:
此外，有两种失败是我们应该关注的：

1. Failure of a Worker Node - Any of the worker nodes running executors can fail, and all in-memory data on those nodes will be lost. If any receivers were running on failed nodes, then their buffered data will be lost.
   工作节点故障-任何运行执行器的工作节点都可能发生故障，这些节点上的所有内存数据都将丢失。如果任何接收器在故障节点上运行，那么它们的缓冲数据将丢失。
2. Failure of the Driver Node - If the driver node running the Spark Streaming application fails, then obviously the SparkContext is lost, and all executors with their in-memory data are lost.
   驱动程序节点的故障-如果运行Spark Streaming应用程序的驱动程序节点发生故障，那么很明显SparkContext会丢失，并且所有执行程序及其内存数据都会丢失。

With this basic knowledge, let us understand the fault-tolerance semantics of Spark Streaming.
有了这些基础知识，让我们理解Spark Streaming的容错语义。

Definitions 定义

The semantics of streaming systems are often captured in terms of how many times each record can be processed by the system. There are three types of guarantees that a system can provide under all possible operating conditions (despite failures, etc.)
流系统的语义通常是根据系统可以处理每个记录的次数来捕获的。在所有可能的操作条件下（尽管有故障等），系统可以提供三种类型的保证。

1. At most once: Each record will be either processed once or not processed at all.
   至多一次：每条记录将被处理一次或根本不处理。
2. At least once: Each record will be processed one or more times. This is stronger than at-most once as it ensure that no data will be lost. But there may be duplicates.
   至少一次：每条记录将被处理一次或多次。这比最多一次更强大，因为它确保不会丢失任何数据。但可能有重复。
3. Exactly once: Each record will be processed exactly once - no data will be lost and no data will be processed multiple times. This is obviously the strongest guarantee of the three.
   精确处理一次：每条记录将被精确处理一次-不会丢失数据，也不会多次处理数据。这显然是三者中最强有力的保障。

Basic Semantics 基本语义

In any stream processing system, broadly speaking, there are three steps in processing the data.
在任何流处理系统中，一般来说，处理数据有三个步骤。

1. Receiving the data: The data is received from sources using Receivers or otherwise.
   接收数据：使用接收器或其他方式从源接收数据。

2. Transforming the data: The received data is transformed using DStream and RDD transformations.
   转换数据：接收到的数据使用DStream和RDD转换进行转换。

3. Pushing out the data: The final transformed data is pushed out to external systems like file systems, databases, dashboards, etc.
   推送数据：最终转换的数据被推送到外部系统，如文件系统，数据库，仪表板等。

If a streaming application has to achieve end-to-end exactly-once guarantees, then each step has to provide an exactly-once guarantee. That is, each record must be received exactly once, transformed exactly once, and pushed to downstream systems exactly once. Let’s understand the semantics of these steps in the context of Spark Streaming.
如果流应用必须实现端到端的精确一次保证，那么每个步骤都必须提供精确一次保证。也就是说，每条记录必须被接收一次，转换一次，并推送到下游系统一次。让我们在Spark Streaming的上下文中理解这些步骤的语义。

1. Receiving the data: Different input sources provide different guarantees. This is discussed in detail in the next subsection.
   接收数据：不同的输入源提供不同的保证。这将在下一小节中详细讨论。

2. Transforming the data: All data that has been received will be processed exactly once, thanks to the guarantees that RDDs provide. Even if there are failures, as long as the received input data is accessible, the final transformed RDDs will always have the same contents.
   转换数据：由于RDD提供的保证，所有接收到的数据都将只处理一次。即使存在故障，只要接收到的输入数据是可访问的，最终转换的RDD将始终具有相同的内容。

3. Pushing out the data: Output operations by default ensure at-least once semantics because it depends on the type of output operation (idempotent, or not) and the semantics of the downstream system (supports transactions or not). But users can implement their own transaction mechanisms to achieve exactly-once semantics. This is discussed in more details later in the section.
   推数据：默认情况下，输出操作确保至少一次语义，因为它取决于输出操作的类型（是否幂等）和下游系统的语义（是否支持事务）。但是用户可以实现自己的事务机制来实现exactly-once语义。这将在本节后面更详细地讨论。

Semantics of Received Data
接收数据的语义

Different input sources provide different guarantees, ranging from at-least once to exactly once. Read for more details.
不同的输入源提供不同的保证，范围从至少一次到恰好一次。请阅读以了解更多详情。

With Files 与文件

If all of the input data is already present in a fault-tolerant file system like HDFS, Spark Streaming can always recover from any failure and process all of the data. This gives exactly-once semantics, meaning all of the data will be processed exactly once no matter what fails.
如果所有输入数据都已经存在于HDFS等容错文件系统中，Spark Streaming可以始终从任何故障中恢复并处理所有数据。这给出了exactly-once语义，意味着无论什么失败，所有数据都将被精确处理一次。

With Receiver-based Sources
基于接收器的源

For input sources based on receivers, the fault-tolerance semantics depend on both the failure scenario and the type of receiver. As we discussed earlier, there are two types of receivers:
对于基于接收器的输入源，容错语义取决于故障场景和接收器类型。如前所述，有两种类型的接收器：

1. Reliable Receiver - These receivers acknowledge reliable sources only after ensuring that the received data has been replicated. If such a receiver fails, the source will not receive acknowledgment for the buffered (unreplicated) data. Therefore, if the receiver is restarted, the source will resend the data, and no data will be lost due to the failure.
   可靠的接收器-这些接收器仅在确保接收到的数据已被复制后才确认可靠的来源。如果这样的接收器失败，源将不会收到缓冲（未复制）数据的确认。因此，如果接收方重启，源端会重新发送数据，不会因为失败而丢失数据。
2. Unreliable Receiver - Such receivers do not send acknowledgment and therefore can lose data when they fail due to worker or driver failures.
   不可靠的接收器-这样的接收器不发送确认，因此当它们由于工作程序或驱动程序故障而失败时可能会丢失数据。

Depending on what type of receivers are used we achieve the following semantics. If a worker node fails, then there is no data loss with reliable receivers. With unreliable receivers, data received but not replicated can get lost. If the driver node fails, then besides these losses, all of the past data that was received and replicated in memory will be lost. This will affect the results of the stateful transformations.
根据使用的接收器类型，我们实现以下语义。如果工作节点发生故障，那么可靠的接收器不会丢失数据。对于不可靠的接收器，接收到但未复制的数据可能会丢失。如果驱动程序节点发生故障，那么除了这些丢失之外，所有在内存中接收和复制的过去数据都将丢失。这将影响有状态转换的结果。

To avoid this loss of past received data, Spark 1.2 introduced write ahead logs which save the received data to fault-tolerant storage. With the write-ahead logs enabled and reliable receivers, there is zero data loss. In terms of semantics, it provides an at-least once guarantee.
为了避免过去接收到的数据丢失，Spark 1.2引入了写前日志，将接收到的数据保存到容错存储中。在启用了预写日志和可靠的接收器的情况下，数据丢失为零。在语义方面，它提供了至少一次保证。

The following table summarizes the semantics under failures:
下表总结了失败下的语义：

With Kafka Direct API
使用Kafka Direct API

In Spark 1.3, we have introduced a new Kafka Direct API, which can ensure that all the Kafka data is received by Spark Streaming exactly once. Along with this, if you implement exactly-once output operation, you can achieve end-to-end exactly-once guarantees. This approach is further discussed in the Kafka Integration Guide.
在Spark 1.3中，我们引入了一个新的Kafka Direct API，它可以确保Spark Streaming只接收一次所有的Kafka数据。沿着，如果您实现了exactly-once输出操作，则可以实现端到端的exactly-once保证。在Kafka集成指南中进一步讨论了这种方法。

Semantics of output operations
输出操作的语义

Output operations (like foreachRDD) have at-least once semantics, that is, the transformed data may get written to an external entity more than once in the event of a worker failure. While this is acceptable for saving to file systems using the saveAs***Files operations (as the file will simply get overwritten with the same data), additional effort may be necessary to achieve exactly-once semantics. There are two approaches.
输出操作（如 foreachRDD ）具有至少一次语义，也就是说，在工作者失败的情况下，转换后的数据可以多次写入外部实体。虽然这对于使用 saveAs***Files 操作保存到文件系统是可以接受的（因为文件将简单地被相同的数据覆盖），但是可能需要额外的努力来实现精确一次语义。有两种方法。

• Idempotent updates: Multiple attempts always write the same data. For example, saveAs***Files always writes the same data to the generated files.
  幂等更新：多次尝试总是写入相同的数据。例如， saveAs***Files 总是将相同的数据写入生成的文件。

• Transactional updates: All updates are made transactionally so that updates are made exactly once atomically. One way to do this would be the following.
  事务更新：所有的更新都是事务性的，所以更新只在原子上进行一次。这样做的一种方法如下。

  • Use the batch time (available in foreachRDD) and the partition index of the RDD to create an identifier. This identifier uniquely identifies a blob data in the streaming application.
    使用批处理时间（在 foreachRDD 中可用）和RDD的分区索引来创建标识符。该标识符唯一地标识流应用中的团块数据。

  • Update external system with this blob transactionally (that is, exactly once, atomically) using the identifier. That is, if the identifier is not already committed, commit the partition data and the identifier atomically. Else, if this was already committed, skip the update.
    使用标识符事务性地（即，原子性地）用这个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
      }
    }
    

Where to Go from Here
从这里往哪里走

• Additional guides  附加指南
  • Kafka Integration Guide  Kafka集成指南
  • Kinesis Integration Guide
    Kinesis集成指南
  • Custom Receiver Guide  定制接收器指南
• Third-party DStream data sources can be found in Third Party Projects
  第三方DStream数据源可以在第三方项目中找到
• API documentation  API文档
  • Scala docs  Scala文档
    • StreamingContext and DStream
      StreamingContext和DStream
    • KafkaUtils, KinesisUtils,
      KafkaUtils，KinesisUtils，
  • Java docs  Java文档
    • JavaStreamingContext, JavaDStream and JavaPairDStream
      JavaStreamingContext、JavaDStream和JavaPairDStream
    • KafkaUtils, KinesisUtils
      KafkaUtils，KinesisUtils
  • Python docs  Python文档
    • StreamingContext and DStream
      StreamingContext和DStream
    • KafkaUtils
• More examples in Scala and Java and Python
  Scala、Java和Python中的更多示例
• Paper and video describing Spark Streaming.
  介绍Spark Streaming的论文和视频。