Spark Streaming 編程指南[中英對照]

基於Spark 2.0 Preview的材料翻譯，原[英]文地址： 

http://spark.apache.org/docs/2.0.0-preview/streaming-programming-guide.html

Streaming應用實戰，參考：http://my.oschina.net/u/2306127/blog/635518

Spark Streaming Programming Guide

Spark Streaming編程指南

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, Flume, 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、Flume、Kinesis、或者TCP sockets等，並且可使用map、reduce、join 和 window等高級接口實現複雜算法的處理。最終，處理的數據能夠被推送到數據庫、文件系統以及動態佈告板。實際上，你還能夠將Spark的機器學習（ machine learning） 和圖計算 （graph processing ）算法用於數據流的處理。

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, Flume, 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, Flume, and Kinesis等源中建立，或者在其它的DStream上執行高級操做。在內部，DStream表明一系列的 RDDs。

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。能夠經過相應的鏈接切換去查看相應語言的代碼。

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.

這在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 socket的數據服務器接收一個文本數據，而後對單詞進行計數。看起來像下面這個樣子。

• Scala
• Java
• Python

First, we import the names of the Spark Streaming classes and some implicit conversions from StreamingContext into our environment in order to add useful methods to other classes we need (like DStream). StreamingContext is the main entry point for all streaming functionality. We create a local StreamingContext with two execution threads, and a batch interval of 1 second.

首先，咱們導入Spark Streaming的類命名空間和一些StreamingContext的轉換工具。 StreamingContext 是全部的Spark Streaming功能的主入口點。咱們建立StreamingContext，指定兩個執行線程和分批間隔爲1秒鐘。

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

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

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

Using this context, we can create a DStream that represents streaming data from a TCP source, specified as hostname (e.g. localhost) and port (e.g. 9999).

使用這個context，咱們能夠建立一個DStream，這是來自於TCP數據源 的流數據，咱們經過hostname (e.g. localhost) 和端口 (e.g. 9999)來指定這個數據源。

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

This lines DStream represents the stream of data that will be received from the data server. Each record in this DStream is a line of text. Next, we want to split the lines by space characters into words.

這裏line是一個DStream對象，表明從服務器收到的流數據。每個DStream中的記錄是一個文本行。下一步，咱們將每一行中以空格分開的單詞分離出來。

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

flatMap is a one-to-many DStream operation that creates a new DStream by generating multiple new records from each record in the source DStream. In this case, each line will be split into multiple words and the stream of words is represented as the words DStream. Next, we want to count these words.

flatMap是「一對多」的DStream操做，經過對源DStream的每個記錄產生多個新的記錄建立新DStream。這裏，每一行將被分解多個單詞，而且單詞流表明了words DStream。下一步，咱們對這些單詞進行計數統計。

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

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

The words DStream is further mapped (one-to-one transformation) to a DStream of (word, 1) pairs, which is then reduced to get the frequency of words in each batch of data. Finally, wordCounts.print() will print a few of the counts generated every second.

words DStream而後映射爲(word, 1)的鍵值對的Dstream，而後用於統計單詞出現的頻度。最後，wordCounts.print()打印出每秒鐘建立出的計數值。

Note that when these lines are executed, Spark Streaming only sets up the computation it will perform when it is started, and no real processing has started yet. To start the processing after all the transformations have been setup, we finally call

注意，上面這些代碼行執行的時候，僅僅是設定了計算執行的邏輯，並無真正的處理數據。在全部的設定完成後，爲了啓動處理，須要調用：

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

The complete code can be found in the Spark Streaming example NetworkWordCount.

完整的代碼能夠Spark Streaming 的例程 NetworkWordCount 中找到。

 

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

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

Scala Java Python # TERMINAL 2: RUNNING NetworkWordCount $ ./bin/run-example streaming.NetworkWordCount localhost 9999 ... ------------------------------------------- Time: 1357008430000 ms ------------------------------------------- (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中心庫訪問。爲了編寫你本身的Spark Streaming程序，您將加入下面的依賴到你的SBT或者Maven工程文件。

• Maven
• SBT

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

For ingesting data from sources like Kafka, Flume, and Kinesis that are not present in the Spark Streaming core API, you will have to add the corresponding artifact spark-streaming-xyz_2.11 to the dependencies. For example, some of the common ones are as follows.

爲了從Kafka/Flume/Kinesis等非Spark Streaming核心API等數據源注入數據，咱們須要添加對應的spark-streaming-xyz_2.11到依賴中。例如，像下面的這樣：

Source	Artifact
Kafka	spark-streaming-kafka-0-8_2.11
Flume	spark-streaming-flume_2.11
Kinesis	spark-streaming-kinesis-asl_2.11 [Amazon Software License]

For an up-to-date list, please refer to the Maven repository for the full list of supported sources and artifacts.

對於最新的列表，參考Maven repository 得到全面的數據源河訪問部件的列表。

---

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 對象必須首先建立做爲總入口。

• Scala
• Java
• Python

A StreamingContext object can be created from a SparkConf object.

 StreamingContext 對象能夠經過 SparkConf 對象建立，以下所示。

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

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

The appName parameter is a name for your application to show on the cluster UI. master is a Spark, Mesos or YARN cluster URL, or a special 「local[*]」 string to run in local mode. In practice, when running on a cluster, you will not want to hardcode master in the program, but rather launch the application with spark-submit and receive it there. However, for local testing and unit tests, you can pass 「local[*]」 to run Spark Streaming in-process (detects the number of cores in the local system). Note that this internally creates a SparkContext (starting point of all Spark functionality) which can be accessed as ssc.sparkContext.

這裏 appName參數是應用在集羣中的名稱。 master 是 Spark, Mesos 或 YARN cluster URL,  或者「local[*]」 字符串指示運行在 local 模式下。實踐中，當運行一個集羣, 您不該該硬編碼 master 參數在集羣中, 而是經過 launch the application with spark-submit 接收其參數。可是, 對於本地測試和單元測試, 你能夠傳遞「local[*]」 來運行 Spark Streaming 在進程內運行(自動檢測本地系統的CPU內核數量)。 注意，這裏內部建立了 SparkContext (全部的Spark 功能的入口點) ，能夠經過 ssc.sparkContext進行存取。

The batch interval must be set based on the latency requirements of your application and available cluster resources. See the Performance Tuning section for more details.

分批間隔時間基於應用延遲需求和可用的集羣資源進行設定（譯註：設定間隔要大於應用數據的最小延遲需求，同時不能設置過小以致於系統沒法在給定的週期內處理完畢），參考 Performance Tuning 部分得到更多信息。

A StreamingContext object can also be created from an existing SparkContext object.

StreamingContext 對象也能夠從已有的 SparkContext 對象中建立。

import org.apache.spark.streaming._

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

After a context is defined, you have to do the following.

1. Define the input sources by creating input DStreams.
2. Define the streaming computations by applying transformation and output operations to DStreams.
3. Start receiving data and processing it using streamingContext.start().
4. Wait for the processing to be stopped (manually or due to any error) using streamingContext.awaitTermination().
5. The processing can be manually stopped using streamingContext.stop().

在context建立以後，能夠接着開始以下的工做：

1. 定義 input sources，經過建立 input DStreams完成。
2. 定義 streaming 計算，經過DStreams的 transformation 和 output 操做實現。
3. 啓動接收數據和處理，經過 streamingContext.start()。
4. 等待處理中止 (一般由於錯誤)，經過streamingContext.awaitTermination().
5. 處理過程能夠手動中止，經過 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.
• stop() on StreamingContext also stops the SparkContext. To stop only the StreamingContext, set the optional parameter of stop() called stopSparkContext to 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.

記住：

• 一旦context啓動, 沒有新的streaming 計算能夠被設置和添加進來。
• 一旦context被中止, 它不能被再次啓動。
• 只有一個StreamingContext在JVM中在同一時間能夠被激活。
• stop() 在StreamingContext執行時，同時中止了SparkContext。爲了僅終止StreamingContext, 在stopSparkContext的Stop時設置選項爲false。
• SparkContext 能夠重用來建立多個 StreamingContexts, 一直到前一個StreamingContext被中止的時候 (不中止 SparkContext) ，才能建立下一個StreamingContext。

---

Discretized Streams (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 Programming Guide）。每一個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轉成words DStream對象，其實做用於lines上的flatMap算子，會施加於lines中的每一個RDD上，並生成新的對應的RDD，而這些新生成的RDD 對象就組成了words這個DStream對象。其過程以下圖所示：

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

輸入DStream和接收器

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.

輸入DStream表明從某種流式數據源流入的數據流。在以前的例子裏，lines 對象就是輸入DStream，它表明從netcat server收到的數據流。每一個輸入DStream（除文件數據流外）都和一個接收器（Receiver – Scala doc, Java doc）相關聯，而接收器則是專門從數據源拉取數據到內存中的對象。

Spark Streaming provides two categories of built-in streaming sources.

• Basic sources: Sources directly available in the StreamingContext API. Examples: file systems, and socket connections.
• Advanced sources: Sources like Kafka, Flume, Kinesis, etc. are available through extra utility classes. These require linking against extra dependencies as discussed in the linking section.

Spark Streaming主要提供兩種內建的流式數據源：

• 基礎數據源（Basic sources）: 在StreamingContext API 中可直接使用的源，如：文件系統，套接字鏈接或者Akka actor。
• 高級數據源（Advanced sources）: 須要依賴額外工具類的源，如：Kafka、Flume、Kinesis、Twitter等數據源。這些數據源都須要增長額外的依賴，詳見依賴連接（linking）這一節。

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對象（詳見後續的性能調優這一小節）。多個DStream對象其實也就同 時建立了多個數據流接收器。可是請注意，Spark的worker/executor 都是長期運行的，所以它們都會各自佔用一個分配給Spark Streaming應用的CPU。因此，在運行Spark Streaming應用的時候，須要注意分配足夠的CPU core（本地運行時，須要足夠的線程）來處理接收到的數據，同時還要足夠的CPU core來運行這些接收器。

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 a input DStream based on a receiver (e.g. sockets, Kafka, Flume, 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).

• 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應用，記得不能將master設爲」local」 或 「local[1]」。這兩個值都只會在本地啓動一個線程。而若是此時你使用一個包含接收器（如：套接字、Kafka、Flume等）的輸入 DStream，那麼這一個線程只能用於運行這個接收器，而處理數據的邏輯就沒有線程來執行了。所以，本地運行時，必定要將master設 爲」local[n]」，其中 n > 接收器的個數（有關master的詳情請參考Spark Properties）。
• 將Spark Streaming應用置於集羣中運行時，一樣，分配給該應用的CPU core數必須大於接收器的總數。不然，該應用就只會接收數據，而不會處理數據。

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鏈接中接收文本數據。而除了TCP套接字外，StreamingContext API 還支持從文件或者Akka actor中拉取數據。

• 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:

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

  Spark Streaming will monitor the directory dataDirectory and process any files created in that directory (files written in nested directories not supported). Note that

  For simple text files, there is an easier method streamingContext.textFileStream(dataDirectory). And file streams do not require running a receiver, hence does not require allocating cores.

  Python API fileStream is not available in the Python API, only textFileStream is available.

  • Scala
  • Java
  • Python
  • The files must have the same data format.
  • The files must be created in the dataDirectory by atomically moving or renaming them into the data directory.
  • Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.

• 文件數據流（File Streams）: 能夠從任何兼容HDFS API（包括：HDFS、S三、NFS等）的文件系統，建立方式以下：

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

  • Spark Streaming將監視該dataDirectory目錄，並處理該目錄下任何新建的文件（目前還不支持嵌套目錄）。注意：

  • 各個文件數據格式必須一致。
  • dataDirectory中的文件必須經過moving或者renaming來建立。
  • 一旦文件move進dataDirectory以後，就不能再改動。因此若是這個文件後續還有寫入，這些新寫入的數據不會被讀取。

  • 另外，文件數據流不是基於接收器的，因此不須要爲其單獨分配一個CPU core。

    Python API fileStream目前暫時不可用，Python目前只支持textFileStream。

    對於簡單的文本文件，更簡單的方式是調用 streamingContext.textFileStream(dataDirectory)。

• Streams based on Custom Receivers: DStreams can be created with data streams received through custom receivers. See the Custom Receiver Guide and DStream Akka for more details.

• 基於自定義Actor的數據流（Streams based on Custom Actors）: DStream能夠由Akka actor建立獲得，只需調用 streamingContext.actorStream(actorProps, actor-name)。詳見自定義接收器（Custom Receiver Guide）。actorStream暫時不支持Python API。

• Queue of RDDs as a Stream: 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.

• RDD隊列數據流（Queue of RDDs as a Stream）: 若是須要測試Spark Streaming應用，你能夠建立一個基於一批RDD的DStream對象，只需調用 streamingContext.queueStream(queueOfRDDs)。RDD會被一個個依次推入隊列，而DStream則會依次以數據 流形式處理這些RDD的數據。

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.

關於套接字、文件以及Akka actor數據流更詳細信息，請參考相關文檔：StreamingContext for Scala,JavaStreamingContext for Java, and StreamingContext for Python。

Advanced Sources

高級數據源

Python API As of Spark 2.0.0, out of these sources, Kafka, Kinesis and Flume are available in the Python API.

Python API 自 Spark 2.0.0（譯註：1.6.1就已經支持了） 起，Kafka、Kinesis、Flume和MQTT這些數據源將支持Python。

This category of sources require interfacing with external non-Spark libraries, some of them with complex dependencies (e.g., Kafka and Flume). 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.

使用這類數據源須要依賴一些額外的代碼庫，有些依賴還挺複雜的（如：Kafka、Flume）。所以爲了減小依賴項版本衝突問題，各個數據源 DStream的相關功能被分割到不一樣的代碼包中，只有用到的時候才須要連接打包進來。

例如，若是你須要使用Twitter的tweets做爲數據源，你 須要如下步驟：

1. Linking: 將spark-streaming-twitter_2.10工件加入到SBT/Maven項目依賴中。
2. Programming: 導入TwitterUtils class，而後調用 TwitterUtils.createStream 建立一個DStream，具體代碼見下放。
3. Deploying: 生成一個uber Jar包，幷包含其全部依賴項（包括 spark-streaming-twitter_2.10及其自身的依賴樹），再部署這個Jar包。部署詳情請參考部署這一節（Deploying section）。

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.

Some of these advanced sources are as follows.

• Kafka: Spark Streaming 2.0.0 is compatible with Kafka 0.8.2.1. See the Kafka Integration Guide for more details.

• Flume: Spark Streaming 2.0.0 is compatible with Flume 1.6.0. See the Flume Integration Guide for more details.

• Kinesis: Spark Streaming 2.0.0 is compatible with Kinesis Client Library 1.2.1. See the Kinesis Integration Guide for more details.

注意，高級數據源在spark-shell中不可用，所以不能用spark-shell來測試基於高級數據源的應用。若是真有須要的話，你須要自行下載相應數據源的Maven工件及其依賴項，並將這些Jar包部署到spark-shell的classpath中。

下面列舉了一些高級數據源：

• Kafka: Spark Streaming 1.6.1 可兼容 Kafka 0.8.2.1。詳見Kafka Integration Guide。
• Flume: Spark Streaming 1.6.1 可兼容 Flume 1.6.0 。詳見Flume Integration Guide。
• Kinesis: Spark Streaming 1.6.1 可兼容 Kinesis Client Library 1.2.1。詳見Kinesis Integration Guide。
• Twitter: Spark Streaming TwitterUtils 使用Twitter4j 經過 Twitter’s Streaming API 拉取公開tweets數據流。認證信息能夠用任何Twitter4j所支持的方法（methods）。你能夠獲取全部的公開數據流，固然也能夠基於某些關鍵詞進行過濾。示例能夠參考TwitterPopularTags 和 TwitterAlgebirdCMS。

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也能夠用自定義的方式建立。你須要作的只是實現一個自定義的接收器（receiver），以便從自定義的數據源接收數據，而後將數據推入Spark中。詳情請參考自定義接收器指南（Custom Receiver Guide）。

Receiver Reliability

接收器可靠性

There can be two kinds of data sources based on their reliability. Sources (like Kafka and Flume) 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:

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

從可靠性角度來劃分，大體有兩種數據源。其中，像Kafka、Flume這樣的數據源，它們支持對所傳輸的數據進行確認。系統收到這類可靠數據源過來的數據，而後發出確認信息，這樣就可以確保任何失敗狀況下，都不會丟數據。所以咱們能夠將接收器也相應地分爲兩類：

1. 可靠接收器（Reliable Receiver） – 可靠接收器會在成功接收並保存好Spark數據副本後，向可靠數據源發送確認信息。
2. 不可靠接收器（Unreliable Receiver） – 不可靠接收器不會發送任何確認信息。不過這種接收器經常使用語於不支持確認的數據源，或者不想引入數據確認的複雜性的數據源。

The details of how to write a reliable receiver are discussed in the Custom Receiver Guide.

自定義接收器指南（Custom Receiver Guide）中詳細討論瞭如何寫一個可靠接收器。

---

Transformations on DStreams

DStream支持的transformation算子

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也支持從輸入DStream通過各類transformation算子映射成新的DStream。DStream支持不少RDD上常見的transformation算子，一些經常使用的見下表：

Transformation	Meaning
map(func)	Return a new DStream by passing each element of the source DStream through a function func.
flatMap(func)	Similar to map, but each input item can be mapped to 0 or more output items.
filter(func)	Return a new DStream by selecting only the records of the source DStream on which func returns true.
repartition(numPartitions)	Changes the level of parallelism in this DStream by creating more or fewer partitions.
union(otherStream)	Return a new DStream that contains the union of the elements in the source DStream and otherDStream.
count()	Return a new DStream of single-element RDDs by counting the number of elements in each RDD of the source DStream.
reduce(func)	Return a new DStream of single-element RDDs by aggregating the elements in each RDD of the source DStream using a function func (which takes two arguments and returns one). The function should be associative and commutative so that it can be computed in parallel.
countByValue()	When called on a DStream of elements of type K, return a new DStream of (K, Long) pairs where the value of each key is its frequency in each RDD of the source DStream.
reduceByKey(func, [numTasks])	When called on a DStream of (K, V) pairs, return a new DStream of (K, V) pairs where the values for each key are aggregated using the given reduce function. Note: By default, this uses Spark's default number of parallel tasks (2 for local mode, and in cluster mode the number is determined by the config property spark.default.parallelism) to do the grouping. You can pass an optional numTasks argument to set a different number of tasks.
join(otherStream, [numTasks])	When called on two DStreams of (K, V) and (K, W) pairs, return a new DStream of (K, (V, W)) pairs with all pairs of elements for each key.
cogroup(otherStream, [numTasks])	When called on a DStream of (K, V) and (K, W) pairs, return a new DStream of (K, Seq[V], Seq[W]) tuples.
transform(func)	Return a new DStream by applying a RDD-to-RDD function to every RDD of the source DStream. This can be used to do arbitrary RDD operations on the DStream.
updateStateByKey(func)	Return a new "state" DStream where the state for each key is updated by applying the given function on the previous state of the key and the new values for the key. This can be used to maintain arbitrary state data for each key.

A few of these transformations are worth discussing in more detail.

Transformation算子	用途
map(func)	返回會一個新的DStream，並將源DStream中每一個元素經過func映射爲新的元素
flatMap(func)	和map相似，不過每一個輸入元素再也不是映射爲一個輸出，而是映射爲0到多個輸出
filter(func)	返回一個新的DStream，幷包含源DStream中被func選中（func返回true）的元素
repartition(numPartitions)	更改DStream的並行度（增長或減小分區數）
union(otherStream)	返回新的DStream，包含源DStream和otherDStream元素的並集
count()	返回一個包含單元素RDDs的DStream，其中每一個元素是源DStream中各個RDD中的元素個數
reduce(func)	返回一個包含單元素RDDs的DStream，其中每一個元素是經過源RDD中各個RDD的元素經func（func輸入兩個參數並返回一個同類型結果數據）聚合獲得的結果。func必須知足結合律，以便支持並行計算。
countByValue()	若是源DStream包含的元素類型爲K，那麼該算子返回新的DStream包含元素爲(K, Long)鍵值對，其中K爲源DStream各個元素，而Long爲該元素出現的次數。
reduceByKey(func, [numTasks])	若是源DStream 包含的元素爲 (K, V) 鍵值對，則該算子返回一個新的也包含(K, V)鍵值對的DStream，其中V是由func聚合獲得的。注意：默認狀況下，該算子使用Spark的默認併發任務數（本地模式爲2，集羣模式下由 spark.default.parallelism 決定）。你能夠經過可選參數numTasks來指定併發任務個數。
join(otherStream, [numTasks])	若是源DStream包含元素爲(K, V)，同時otherDStream包含元素爲(K, W)鍵值對，則該算子返回一個新的DStream，其中源DStream和otherDStream中每一個K都對應一個 (K, (V, W))鍵值對元素。
cogroup(otherStream, [numTasks])	若是源DStream包含元素爲(K, V)，同時otherDStream包含元素爲(K, W)鍵值對，則該算子返回一個新的DStream，其中每一個元素類型爲包含(K, Seq[V], Seq[W])的tuple。
transform(func)	返回一個新的DStream，其包含的RDD爲源RDD通過func操做後獲得的結果。利用該算子能夠對DStream施加任意的操做。
updateStateByKey(func)	返回一個包含新」狀態」的DStream。源DStream中每一個key及其對應的values會做爲func的輸入，而func能夠用於對每一個key的「狀態」數據做任意的更新操做。

下面咱們會挑幾個transformation算子深刻討論一下。

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.

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.

updateStateByKey 算子支持維護一個任意的狀態。要實現這一點，只須要兩步：

1. 定義狀態 – 狀態數據能夠是任意類型。
2. 定義狀態更新函數 – 定義好一個函數，其輸入爲數據流以前的狀態和新的數據流數據，且可其更新步驟1中定義的輸入數據流的狀態。

在每個批次數據到達後，Spark都會調用狀態更新函數，來更新全部已有key（無論key是否存在於本批次中）的狀態。若是狀態更新函數返回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:

舉例以下。假設你須要維護一個流式應用，統計數據流中每一個單詞的出現次數。這裏將各個單詞的出現次數這個整型數定義爲狀態。咱們接下來定義狀態更新函數以下：

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

This is applied on a DStream containing words (say, the pairs DStream containing (word, 1) pairs in the earlier example).

該狀態更新函數能夠做用於一個包括(word, 1) 鍵值對的DStream上（見本文開頭的例子）。

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

The update function will be called for each word, with newValues having a sequence of 1’s (from the (word, 1) pairs) and the runningCount having the previous count.

該狀態更新函數會爲每一個單詞調用一次，且相應的newValues是一個包含不少個」1″的數組（這些1來自於(word,1)鍵值對），而runningCount包含以前該單詞的計數。本例的完整代碼請參考 StatefulNetworkWordCount.scala。

Note that using updateStateByKey requires the checkpoint directory to be configured, which is discussed in detail in the checkpointing section.

注意，調用updateStateByKey前須要配置檢查點目錄，後續對此有詳細的討論，見檢查點（checkpointing）這節。

Transform Operation

transform算子

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）能夠支持任意的RDD到RDD的映射操做。也就是說，你能夠用tranform算子來包裝 任何DStream API所不支持的RDD算子。例如，將DStream每一個批次中的RDD和另外一個Dataset進行關聯（join）操做，這個功能DStream API並無直接支持。不過你能夠用transform來實現這個功能，可見transform其實爲DStream提供了很是強大的功能支持。好比說， 你能夠用事先算好的垃圾信息，對DStream進行實時過濾。

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

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.

注意，這裏transform包含的算子，其調用時間間隔和批次間隔是相同的。因此你能夠基於時間改變對RDD的操做，如：在不一樣批次，調用不一樣的RDD算子，設置不一樣的RDD分區或者廣播變量等。

Window Operations

基於窗口（window）的算子

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一樣也提供基於時間窗口的計算，也就是說，你能夠對某一個滑動時間窗內的數據施加特定tranformation算子。以下圖所示：

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.

• window length - The duration of the window (3 in the figure).
• sliding interval - The interval at which the window operation is performed (2 in the figure).

如上圖所示，每次窗口滑動時，源DStream中落入窗口的RDDs就會被合併成新的windowed DStream。在上圖的例子中，這個操做會施加於3個RDD單元，而滑動距離是2個RDD單元。由此能夠得出任何窗口相關操做都須要指定一下兩個參數：

• （窗口長度）window length – 窗口覆蓋的時間長度（上圖中爲3）
• （滑動距離）sliding interval – 窗口啓動的時間間隔（上圖中爲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秒內的單詞計數。爲此，咱們須要在包含(word, 1)鍵值對的DStream上，對最近30秒的數據調用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))

Some of the common window operations are as follows. All of these operations take the said two parameters - windowLength and slideInterval.

Transformation	Meaning
window(windowLength, slideInterval)	Return a new DStream which is computed based on windowed batches of the source DStream.
countByWindow(windowLength, slideInterval)	Return a sliding window count of elements in the stream.
reduceByWindow(func, windowLength, slideInterval)	Return a new single-element stream, created by aggregating elements in the stream over a sliding interval using func. The function should be associative and commutative so that it can be computed correctly in parallel.
reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])	When called on a DStream of (K, V) pairs, returns a new DStream of (K, V) pairs where the values for each key are aggregated using the given reduce function func over batches in a sliding window. Note: By default, this uses Spark's default number of parallel tasks (2 for local mode, and in cluster mode the number is determined by the config property spark.default.parallelism) to do the grouping. You can pass an optional numTasks argument to set a different number of tasks.
reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])	A more efficient version of the above reduceByKeyAndWindow() where the reduce value of each window is calculated incrementally using the reduce values of the previous window. This is done by reducing the new data that enters the sliding window, and 「inverse reducing」 the old data that leaves the window. An example would be that of 「adding」 and 「subtracting」 counts of keys as the window slides. However, it is applicable only to 「invertible reduce functions」, that is, those reduce functions which have a corresponding 「inverse reduce」 function (taken as parameter invFunc). Like in reduceByKeyAndWindow, the number of reduce tasks is configurable through an optional argument. Note that checkpointing must be enabled for using this operation.
countByValueAndWindow(windowLength, slideInterval, [numTasks])	When called on a DStream of (K, V) pairs, returns a new DStream of (K, Long) pairs where the value of each key is its frequency within a sliding window. Like in reduceByKeyAndWindow, the number of reduce tasks is configurable through an optional argument.

如下列出了經常使用的窗口算子。全部這些算子都有前面提到的那兩個參數 – 窗口長度 和 滑動距離。

Transformation窗口算子	用途
window(windowLength, slideInterval)	將源DStream窗口化，並返回轉化後的DStream
countByWindow(windowLength,slideInterval)	返回數據流在一個滑動窗口內的元素個數
reduceByWindow(func, windowLength,slideInterval)	基於數據流在一個滑動窗口內的元素，用func作聚合，返回一個單元素數據流。func必須知足結合律，以便支持並行計算。
reduceByKeyAndWindow(func,windowLength, slideInterval, [numTasks])	基於(K, V)鍵值對DStream，將一個滑動窗口內的數據進行聚合，返回一個新的包含(K,V)鍵值對的DStream，其中每一個value都是各個key通過func聚合後的結果。 注意：若是不指定numTasks，其值將使用Spark的默認並行任務數（本地模式下爲2，集羣模式下由 spark.default.parallelism決定）。固然，你也能夠經過numTasks來指定任務個數。
reduceByKeyAndWindow(func, invFunc,windowLength,slideInterval, [numTasks])	和前面的reduceByKeyAndWindow() 相似，只是這個版本會用以前滑動窗口計算結果，遞增地計算每一個窗口的歸約結果。當新的數據進入窗口時，這些values會被輸入func作歸約計算，而這 些數據離開窗口時，對應的這些values又會被輸入 invFunc 作」反歸約」計算。舉個簡單的例子，就是把新進入窗口數據中各個單詞個數「增長」到各個單詞統計結果上，同時把離開窗口數據中各個單詞的統計個數從相應的 統計結果中「減掉」。不過，你的本身定義好」反歸約」函數，即：該算子不只有歸約函數（見參數func），還得有一個對應的」反歸約」函數（見參數中的 invFunc）。和前面的reduceByKeyAndWindow() 相似，該算子也有一個可選參數numTasks來指定並行任務數。注意，這個算子須要配置好檢查點（checkpointing）才能用。
countByValueAndWindow(windowLength,slideInterval, [numTasks])	基於包含(K, V)鍵值對的DStream，返回新的包含(K, Long)鍵值對的DStream。其中的Long value都是滑動窗口內key出現次數的計數。 和前面的reduceByKeyAndWindow() 相似，該算子也有一個可選參數numTasks來指定並行任務數。

Join Operations

Join相關算子

Finally, its worth highlighting how easily you can perform different kinds of joins in Spark Streaming.

最後，值得一提的是，你在Spark Streaming中作各類關聯（join）操做很是簡單。

Stream-stream joins

流-流（Stream-stream）關聯

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)

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進行join。一樣，你能夠相似地使用 leftOuterJoin, rightOuterJoin, fullOuterJoin 等。此外，你還能夠基於窗口來join不一樣的數據流，其實現也很簡單，以下；）

• Scala
• Java
• Python

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

Stream-dataset joins

流-數據集（stream-dataset）關聯

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

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.

實際上，在上面代碼裏，你能夠動態地該表join的數據集（dataset）。傳給tranform算子的操做函數會在每一個批次從新求值，因此每次該函數都會用最新的dataset值，因此不一樣批次間你能夠改變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 transformation算子列表見API文檔。Scala請參考 DStream 和 PairDStreamFunctions. Java請參考 JavaDStream 和 JavaPairDStream. Python見 DStream。

---

---

Output Operations on DStreams

DStream輸出算子

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:

Output Operation	Meaning
print()	Prints the first ten elements of every batch of data in a DStream on the driver node running the streaming application. This is useful for development and debugging. Python API This is called pprint() in the Python API.
saveAsTextFiles(prefix, [suffix])	Save this DStream's contents as text files. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]".
saveAsObjectFiles(prefix, [suffix])	Save this DStream's contents as SequenceFiles of serialized Java objects. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]". Python API This is not available in the Python API.
saveAsHadoopFiles(prefix, [suffix])	Save this DStream's contents as Hadoop files. The file name at each batch interval is generated based on prefix and suffix: "prefix-TIME_IN_MS[.suffix]". Python API This is not available in the Python API.
foreachRDD(func)	The most generic output operator that applies a function, func, to each RDD generated from the stream. This function should push the data in each RDD to an external system, such as saving the RDD to files, or writing it over the network to a database. Note that the function func is executed in the driver process running the streaming application, and will usually have RDD actions in it that will force the computation of the streaming RDDs.

輸出算子能夠將DStream的數據推送到外部系統，如：數據庫或者文件系統。由於輸出算子會將最終完成轉換的數據輸出到外部系統，所以只有輸出算 子調用時，纔會真正觸發DStream transformation算子的真正執行（這一點相似於RDD 的action算子）。目前所支持的輸出算子以下表：

輸出算子	用途
print()	在驅動器（driver）節點上打印DStream每一個批次中的頭十個元素。 Python API 對應的Python API爲 pprint()
saveAsTextFiles(prefix, [suffix])	將DStream的內容保存到文本文件。 每一個批次一個文件，各文件命名規則爲 「prefix-TIME_IN_MS[.suffix]」
saveAsObjectFiles(prefix, [suffix])	將DStream內容以序列化Java對象的形式保存到順序文件中。 每一個批次一個文件，各文件命名規則爲 「prefix-TIME_IN_MS[.suffix]」Python API 暫不支持Python
saveAsHadoopFiles(prefix, [suffix])	將DStream內容保存到Hadoop文件中。 每一個批次一個文件，各文件命名規則爲 「prefix-TIME_IN_MS[.suffix]」Python API 暫不支持Python
foreachRDD(func)	這是最通用的輸出算子了，該算子接收一個函數func，func將做用於DStream的每一個RDD上。 func應該實現將每一個RDD的數據推到外部系統中，好比：保存到文件或者寫到數據庫中。 注意，func函數是在streaming應用的驅動器進程中執行的，因此若是其中包含RDD的action算子，就會觸發對DStream中RDDs的實際計算過程。

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是一個很是強大的原生工具函數，用戶能夠基於此算子將DStream數據推送到外部系統中。不過用戶須要瞭解如何正確而高效地使用這個工具。如下列舉了一些常見的錯誤。

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

一般，對外部系統寫入數據須要一些鏈接對象（如：遠程server的TCP鏈接），以便發送數據給遠程系統。所以，開發人員可能會不經意地在Spark驅動器（driver）進程中建立一個鏈接對象，而後又試圖在Spark worker節點上使用這個鏈接。以下例所示：

• Scala
• Python

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

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

這段代碼是錯誤的，由於它須要把鏈接對象序列化，再從驅動器節點發送到worker節點。而這些鏈接對象一般都是不能跨節點（機器）傳遞的。好比，鏈接對 象一般都不能序列化，或者在另外一個進程中反序列化後再次初始化（鏈接對象一般都須要初始化，所以從驅動節點發到worker節點後可能須要從新初始化） 等。解決此類錯誤的辦法就是在worker節點上建立鏈接對象。

However, this can lead to another common mistake - creating a new connection for every record. For example,

然而，有些開發人員可能會走到另外一個極端 – 爲每條記錄都建立一個鏈接對象，例如：

• Scala
• Python

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

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
• Python

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

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批次之間複用鏈接對象。開發者能夠維護一個靜態鏈接池來保存鏈接對象，以便在不一樣批次的多個RDD之間共享同一組鏈接對象，示例以下：

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

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.

• By default, output operations are executed one-at-a-time. And they are executed in the order they are defined in the application.

其餘要點:

• DStream的轉化執行也是懶惰的，須要輸出算子來觸發，這一點和RDD的懶惰執行由action算子觸發很相似。特別地，DStream輸出 算子中包含的RDD action算子會強制觸發對所接收數據的處理。所以，若是你的Streaming應用中沒有輸出算子，或者你用了 dstream.foreachRDD(func)卻沒有在func中調用RDD action算子，那麼這個應用只會接收數據，而不會處理數據，接收到的數據最後只是被簡單地丟棄掉了。
• 默認地，輸出算子只能一次執行一個，且按照它們在應用程序代碼中定義的順序執行。

---

Accumulators and Broadcast Variables

累加器和廣播變量

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.

首先須要注意的是，累加器（Accumulators）和廣播變量（Broadcast variables）是沒法從Spark Streaming的檢查點中恢復回來的。因此若是你開啓了檢查點功能，並同時在使用累加器和廣播變量，那麼你最好是使用懶惰實例化的單例模式，由於這樣累加器和廣播變量才能在驅動器（driver）故障恢復後從新實例化。代碼示例以下：

• Scala
• Java
• Python

object WordBlacklist {

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

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

object DroppedWordsCounter {

  @volatile private var instance: Accumulator[Long] = null

  def getInstance(sc: SparkContext): Accumulator[Long] = {
    if (instance == null) {
      synchronized {
        if (instance == null) {
          instance = sc.accumulator(0L, "WordsInBlacklistCounter")
        }
      }
    }
    instance
  }
}

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

See the full source code.

這裏有完整代碼：source code。

---

DataFrame and SQL Operations

DataFrame和SQL相關算子

You can easily use DataFrames and SQL operations on streaming data. You have to create a SQLContext 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 SQLContext. 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.

在Streaming應用中能夠調用DataFrames and SQL來 處理流式數據。開發者能夠用經過StreamingContext中的SparkContext對象來建立一個SQLContext，而且，開發者須要確保一旦驅動器（driver）故障恢復後，該SQLContext對象能從新建立出來。一樣，你仍是可使用懶惰建立的單例模式來實例化 SQLContext，以下面的代碼所示，這裏咱們將最開始的那個例子作了一些修改，使用DataFrame和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 SQLContext
  val sqlContext = SQLContext.getOrCreate(rdd.sparkContext)
  import sqlContext.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 = 
    sqlContext.sql("select word, count(*) as total from words group by word")
  wordCountsDataFrame.show()
}

See the full source code.

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

See the DataFrames and SQL guide to learn more about DataFrames.

這裏有完整代碼：source code。

你也能夠在其餘線程裏執行SQL查詢（異步查詢，即：執行SQL查詢的線程和運行StreamingContext的線程不一樣）。不過這種狀況下， 你須要確保查詢的時候 StreamingContext 沒有把所需的數據丟棄掉，不然StreamingContext有可能已將老的RDD數據丟棄掉了，那麼異步查詢的SQL語句也可能沒法獲得查詢結果。舉 個栗子，若是你須要查詢上一個批次的數據，可是你的SQL查詢可能要執行5分鐘，那麼你就須要StreamingContext至少保留最近5分鐘的數 據：streamingContext.remember(Minutes(5)) （這是Scala爲例，其餘語言差很少）

更多DataFrame和SQL的文檔見這裏： DataFrames and SQL

---

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 提供了不少機器學習算法。首先，你須要關注的是流式計算相關的機器學習算法（如：Streaming Linear Regression, Streaming 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().

For input streams that receive data over the network (such as, Kafka, Flume, sockets, etc.), the default persistence level is set to replicate the data to two nodes for fault-tolerance.

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相似，DStream也支持將數據持久化到內存中。只須要調用 DStream的persist() 方法，該方法內部會自動調用DStream中每一個RDD的persist方法進而將數據持久化到內存中。這對於可能須要計算不少次的DStream很是有 用（例如：對於同一個批數據調用多個算子）。對於基於滑動窗口的算子，如：reduceByWindow和reduceByKeyAndWindow，或 者有狀態的算子，如：updateStateByKey，數據持久化就更重要了。所以，滑動窗口算子產生的DStream對象默認會自動持久化到內存中 （不須要開發者調用persist）。

對於從網絡接收數據的輸入數據流（如：Kafka、Flume、socket等），默認的持久化級別會將數據持久化到兩個不一樣的節點上互爲備份副本，以便支持容錯。

注意，與RDD不一樣的是，DStream的默認持久化級別是將數據序列化到內存中。進一步的討論見性能調優這一小節。關於持久化級別（或者存儲級別）的更詳細說明見Spark編程指南（Spark Programming Guide）。

---

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.

• 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:
  • Configuration - The configuration that was used to create the streaming application.
  • DStream operations - The set of DStream operations that define the streaming application.
  • 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.

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.

通常來講Streaming 應用都須要7*24小時長期運行，因此必須對一些與業務邏輯無關的故障有很好的容錯（如：系統故障、JVM崩潰等）。對於這些可能性，Spark Streaming 必須在檢查點保存足夠的信息到一些可容錯的外部存儲系統中，以便可以隨時從故障中恢復回來。因此，檢查點須要保存如下兩種數據：

• 元數據檢查點（Metadata checkpointing） – 保存流式計算邏輯的定義信息到外部可容錯存儲系統（如：HDFS）。主要用途是用於在故障後回覆應用程序自己（後續詳談）。元數包括：
  • Configuration – 建立Streaming應用程序的配置信息。
  • DStream operations – 定義流式處理邏輯的DStream操做信息。
  • Incomplete batches – 已經排隊但未處理完的批次信息。
• 數據檢查點（Data checkpointing） – 將生成的RDD保存到可靠的存儲中。這對一些須要跨批次組合數據或者有狀態的算子來講頗有必要。在這種轉換算子中，每每新生成的RDD是依賴於前幾個批次 的RDD，所以隨着時間的推移，有可能產生很長的依賴鏈條。爲了不在恢復數據的時候須要恢復整個依賴鏈條上全部的數據，檢查點須要週期性地保存一些中間 RDD狀態信息，以斬斷無限制增加的依賴鏈條和恢復時間。

總之，元數據檢查點主要是爲了恢復驅動器節點上的故障，而數據或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.
• 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.

什麼時候啓用檢查點

若是有如下狀況出現，你就必須啓用檢查點了：

• 使用了有狀態的轉換算子（Usage of stateful transformations） – 無論是用了 updateStateByKey 仍是用了 reduceByKeyAndWindow（有」反歸約」函數的那個版本），你都必須配置檢查點目錄來週期性地保存RDD檢查點。
• 支持驅動器故障中恢復（Recovering from failures of the driver running the application） – 這時候須要元數據檢查點以便恢復流式處理的進度信息。

注意，一些簡單的流式應用，若是沒有用到前面所說的有狀態轉換算子，則徹底能夠不開啓檢查點。不過這樣的話，驅動器（driver）故障恢復後，有 可能會丟失部分數據（有些已經接收但還未處理的數據可能會丟失）。不過一般這點丟失時可接受的，不少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.

• 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().
• When the program is being restarted after failure, it will re-create a StreamingContext from the checkpoint data in the checkpoint directory.

• Scala
• Java
• Python

This behavior is made simple by using StreamingContext.getOrCreate. This is used as follows.

如何配置檢查點

檢查點的啓用，只須要設置好保存檢查點信息的檢查點目錄便可，通常會會將這個目錄設爲一些可容錯的、可靠性較高的文件系統（如：HDFS、S3 等）。開發者只須要調用 streamingContext.checkpoint(checkpointDirectory)。設置好檢查點，你就可使用前面提到的有狀態轉換 算子了。另外，若是你須要你的應用可以支持從驅動器故障中恢復，你可能須要重寫部分代碼，實現如下行爲：

• 若是程序是首次啓動，就須要new一個新的StreamingContext，並定義好全部的數據流處理，而後調用StreamingContext.start()。
• 若是程序是故障後重啓，就須要從檢查點目錄中的數據中從新構建StreamingContext對象。
• Scala
• Java
• Python

不過這個行爲能夠用StreamingContext.getOrCreate來實現，示例以下：

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

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

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

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

If the checkpointDirectory exists, then the context will be recreated from the checkpoint data. If the directory does not exist (i.e., running for the first time), then the function functionToCreateContext will be called to create a new context and set up the DStreams. See the Scala example RecoverableNetworkWordCount. This example appends the word counts of network data into a file.

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.

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.

若是 checkpointDirectory 目錄存在，則context對象會從檢查點數據從新構建出來。若是該目錄不存在（如：首次運行），則 functionToCreateContext 函數會被調用，建立一個新的StreamingContext對象並定義好DStream數據流。完整的示例請參見RecoverableNetworkWordCount，這個例子會將網絡數據中的單詞計數統計結果添加到一個文件中。

除了使用getOrCreate以外，開發者還須要確保驅動器進程能在故障後重啓。這一點只能由應用的部署環境基礎設施來保證。進一步的討論見部署（Deployment）這一節。

另外須要注意的是，RDD檢查點會增長額外的保存數據的開銷。這可能會致使數據流的處理時間變長。所以，你必須仔細的調整檢查點間隔時間。若是批次 間隔過小（好比：1秒），那麼對每一個批次保存檢查點數據將大大減少吞吐量。另外一方面，檢查點保存過於頻繁又會致使血統信息和任務個數的增長，這一樣會影響 系統性能。對於須要RDD檢查點的有狀態轉換算子，默認的間隔是批次間隔的整數倍，且最小10秒。開發人員能夠這樣來自定義這個間 隔：dstream.checkpoint(checkpointInterval)。通常推薦設爲批次間隔時間的5~10倍。

---

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.

• Cluster with a cluster manager - This is the general requirement of any Spark application, and discussed in detail in the deployment guide.

• 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, Flume), 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-8_2.11 and all its transitive dependencies in the application JAR.

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

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

• 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.
  • YARN - Yarn supports a similar mechanism for automatically restarting an application. Please refer to YARN documentation for more details.
  • Mesos - Marathon has been used to achieve this with 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.

• 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 應用，你首先須要具有如下條件：

• 集羣以及集羣管理器 – 這是通常Spark應用的基本要求，詳見 deployment guide。
• 給Spark應用打個JAR包 – 你須要將你的應用打成一個JAR包。若是使用spark-submit 提交應用，那麼你不須要提供Spark和Spark Streaming的相關JAR包。可是，若是你使用了高級數據源（advanced sources – 如：Kafka、Flume、Twitter等），那麼你須要將這些高級數據源相關的JAR包及其依賴一塊兒打包並部署。例如，若是你使用了 TwitterUtils，那麼就必須將spark-streaming-twitter_2.10及其相關依賴都打到應用的JAR包中。
• 爲執行器（executor）預留足夠的內存 – 執行器必須配置預留好足夠的內存，由於接受到的數據都得存在內存裏。注意，若是某些窗口長度達到10分鐘，那也就是說你的系統必須知道保留10分鐘的數據在內存裏。可見，到底預留多少內存是取決於你的應用處理邏輯的。
• 配置檢查點 – 若是你的流式應用須要檢查點，那麼你須要配置一個Hadoop API兼容的可容錯存儲目錄做爲檢查點目錄，流式應用的信息會寫入這個目錄，故障恢復時會用到這個目錄下的數據。詳見前面的檢查點小節。
• 配置驅動程序自動重啓 – 流式應用自動恢復的前提就是，部署基礎設施可以監控驅動器進程，而且可以在其故障時，自動重啓之。不一樣的集羣管理器有不一樣的工具來實現這一功能：
  • Spark獨立部署 – Spark獨立部署集羣能夠支持將Spark應用的驅動器提交到集羣的某個worker節點上運行。同時，Spark的集羣管理器能夠對該驅動器進程進行 監控，一旦驅動器退出且返回非0值，或者因worker節點原始失敗，Spark集羣管理器將自動重啓這個驅動器。詳見Spark獨立部署指南（Spark Standalone guide）。
  • YARN – YARN支持和獨立部署相似的重啓機制。詳細請參考YARN的文檔。
  • Mesos – Mesos上須要用Marathon來實現這一功能。
• 配置WAL（write ahead log）- 從Spark 1.2起，咱們引入了write ahead log來提升容錯性。若是啓用這個功能，則全部接收到的數據都會以write ahead log形式寫入配置好的檢查點目錄中。這樣就能確保數據零丟失（容錯語義有詳細的討論）。用戶只需將 spark.streaming.receiver.writeAheadLog 設爲true。不過，這一樣可能會致使接收器的吞吐量降低。不過你能夠啓動多個接收器並行接收數據，從而提高總體的吞吐量（more receivers in parallel）。 另外，建議在啓用WAL後禁用掉接收數據多副本功能，由於WAL其實已是存儲在一個多副本存儲系統中了。你只須要把存儲級別設爲 StorageLevel.MEMORY_AND_DISK_SER。若是是使用S3（或者其餘不支持flushing的文件系統）存儲WAL，必定要記 得啓用這兩個標識：spark.streaming.driver.writeAheadLog.closeFileAfterWrite 和 spark.streaming.receiver.writeAheadLog.closeFileAfterWrite。更詳細請參考： Spark Streaming Configuration。
• 設置好最大接收速率 – 若是集羣可用資源不足以跟上接收數據的速度，那麼能夠在接收器設置一下最大接收速率，即：每秒接收記錄的條數。相關的主要配置 有：spark.streaming.receiver.maxRate，若是使用Kafka Direct API 還須要設置 spark.streaming.kafka.maxRatePerPartition。從Spark 1.5起，咱們引入了backpressure的概念來動態地根據集羣處理速度，評估並調整該接收速率。用戶只需將 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.

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

• 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, and Flume) 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.

升級Spark Streaming應用程序代碼，可使用如下兩種方式：

• 新的Streaming程序和老的並行跑一段時間，新程序完成初始化之後，再關閉老的。注意，這種方式適用於能同時發送數據到多個目標的數據源（即：數據源同時將數據發給新老兩個Streaming應用程序）。
• 老程序可以優雅地退出（參考  StreamingContext.stop(...) or JavaStreamingContext.stop(...) ）， 即：確保所收到的數據都已經處理完畢後再退出。而後再啓動新的Streaming程序，而新程序將接着在老程序退出點上繼續拉取數據。注意，這種方式須要 數據源支持數據緩存（或者叫數據堆積，如：Kafka、Flume），由於在新舊程序交接的這個空檔時間，數據須要在數據源處緩存。目前還不能支持從檢查 點重啓，由於檢查點存儲的信息包含老程序中的序列化對象信息，在新程序中將其反序列化可能會出錯。這種狀況下，只能要麼指定一個新的檢查點目錄，要麼刪除 老的檢查點目錄。

---

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.

The following two metrics in web UI are particularly important:

• 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自身的監控能力（monitoring capabilities）以外，對Spark Streaming還有一些額外的監控功能可用。若是實例化了StreamingContext，那麼你能夠在Spark web UI上看到多出了一個Streaming tab頁，上面顯示了正在運行的接收器（是否活躍，接收記錄的條數，失敗信息等）和處理完的批次信息（批次處理時間，查詢延時等）。這些信息均可以用來監控streaming應用。

web UI上有兩個度量特別重要：

• 批次處理耗時（Processing Time） – 處理單個批次耗時
• 批次調度延時（Scheduling Delay） -各批次在隊列中等待時間（等待上一個批次處理完）

若是批次處理耗時一直比批次間隔時間大，或者批次調度延時持續上升，就意味着系統處理速度跟不上數據接收速度。這時候你就得考慮一下怎麼把批次處理時間降下來（reducing）。

Spark Streaming程序的處理進度能夠用StreamingListener接口來監聽，這個接口能夠監聽到接收器的狀態和處理時間。不過須要注意的是，這是一個developer 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:

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

性能調優

要得到Spark Streaming應用的最佳性能須要一點點調優工做。本節將深刻解釋一些可以改進Streaming應用性能的配置和參數。整體上來講，你須要考慮這兩方面的事情：

1. 提升集羣資源利用率，減小單批次處理耗時。
2. 設置合適的批次大小，以便使數據處理速度能跟上數據接收速度。

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對每一個批次的處理時間。細節將在優化指南（Tuning Guide）中詳談。這裏僅列舉一些最重要的。

Level of Parallelism in Data Receiving

Receiving data over the network (like Kafka, Flume, 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、Flume、socket等接收數據）須要在Spark中序列化並存儲數據。

若是接收數據的過程是系統瓶頸，那麼能夠考慮增長數據接收的並行度。注意，每一個輸入DStream只包含一個單獨的接收器（receiver，運行 約worker節點），每一個接收器單獨接收一路數據流。因此，配置多個輸入DStream就能從數據源的不一樣分區分別接收多個數據流。例如，能夠將從 Kafka拉取兩個topic的數據流分紅兩個Kafka輸入數據流，每一個數據流拉取其中一個topic的數據，這樣一來會同時有兩個接收器並行地接收數 據，於是增長了整體的吞吐量。同時，另外一方面咱們又能夠把這些DStream數據流合併成一個，而後能夠在合併後的DStream上使用任何可用的 transformation算子。示例代碼以下：

• Scala
• Java
• Python

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

Another parameter that should be considered is the receiver’s blocking 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.

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.

另外一個能夠考慮優化的參數就是接收器的阻塞間隔，該參數由配置參數（configuration parameter）spark.streaming.blockInterval 決定。大多數接收器都會將數據合併成一個個數據塊，而後再保存到spark內存中。對於map類算子來講，每一個批次中數據塊的個數將會決定處理這批數據並 行任務的個數，每一個接收器每批次數據處理任務數約等於 （批次間隔 / 數據塊間隔）。例如，對於2秒的批次間隔，若是數據塊間隔爲200ms，則建立的併發任務數爲10。若是任務數太少（少於單機cpu core個數），則資源利用不夠充分。如需增長這個任務數，對於給定的批次間隔來講，只須要減小數據塊間隔便可。不過，咱們仍是建議數據塊間隔至少要 50ms，不然任務的啓動開銷佔比就過高了。

另外一個切分接收數據流的方法是，顯示地將輸入數據流劃分爲多個分區（使用 inputStream.repartition(<number of partitions>)）。該操做會在處理前，將數據散開從新分發到集羣中多個節點上。

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.

在計算各個階段（stage）中，任何一個階段的併發任務數不足都有可能形成集羣資源利用率低。例如，對於reduce類的算子， 如：reduceByKey 和 reduceByKeyAndWindow，其默認的併發任務數是由 spark.default.parallelism 決定的。你既能夠修改這個默認值（spark.default.parallelism），也能夠經過參數指定這個併發數量（見PairDStreamFunctions）。

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.

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

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

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.

調整數據的序列化格式能夠大大減小數據序列化的開銷。在spark Streaming中主要有兩種類型的數據須要序列化：

• 輸入數據: 默認地，接收器收到的數據是以 StorageLevel.MEMORY_AND_DISK_SER_2 的 存儲級別存儲到執行器（executor）內存中的。也就是說，收到的數據會被序列化以減小GC開銷，同時保存兩個副本以容錯。同時，數據會優先保存在內 存裏，當內存不足時才吐出到磁盤上。很明顯，這個過程當中會有數據序列化的開銷 – 接收器首先將收到的數據反序列化，而後再以spark所配置指定的格式來序列化數據。
• Streaming算子所生產的持久化的RDDs: Streaming計算所生成的RDD可能會持久化到內存中。例如，基於窗口的算子會將數據持久化到內存，由於窗口數據可能會屢次處理。所不一樣的是，spark core默認用 StorageLevel.MEMORY_ONLY 級別持久化RDD數據，而spark streaming默認使用StorageLevel.MEMORY_ONLY_SER 級別持久化接收到的數據，以便儘可能減小GC開銷。

無論是上面哪種數據，均可以使用Kryo序列化來減小CPU和內存開銷，詳見Spark Tuning Guide。另，對於Kryo，你能夠考慮這些優化：註冊自定義類型，禁用對象引用跟蹤（詳見Configuration Guide）。

在一些特定的場景下，若是數據量不是很大，那麼你能夠考慮不用序列化格式，不過你須要注意的是取消序列化是否會致使大量的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 slaves may be significant and will make it hard to achieve sub-second latencies. The overhead can be reduced by the following changes:

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

These changes may reduce batch processing time by 100s of milliseconds, thus allowing sub-second batch size to be viable.

任務啓動開銷

若是每秒啓動的任務數過多（好比每秒50個以上），那麼將任務發送給slave節點的開銷會明顯增長，那麼你也就很難達到亞秒級（sub-second）的延遲。不過如下兩個方法能夠減小任務的啓動開銷：

• 任務序列化（Task Serialization）: 使用Kryo來序列化任務，以減小任務自己的大小，從而提升發送任務的速度。任務的序列化格式是由 spark.closure.serializer 屬性決定的。不過，目前還不支持閉包序列化，將來的版本可能會增長對此的支持。
• 執行模式（Execution mode）: Spark獨立部署或者Mesos粗粒度模式下任務的啓動時間比Mesos細粒度模式下的任務啓動時間要短。詳見Running on Mesos guide。

這些調整有可能可以減小100ms的批次處理時間，這也使得亞秒級的批次間隔成爲可能。

---

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.

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.

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

設置合適的批次間隔

要想streaming應用在集羣上穩定運行，那麼系統處理數據的速度必須能跟上其接收數據的速度。換句話說，批次數據的處理速度應該和其生成速度同樣快。對於特定的應用來講，能夠從其對應的監控（monitoring）頁面上觀察驗證，頁面上顯示的處理耗時應該要小於批次間隔時間。

根據spark streaming計算的性質，在必定的集羣資源限制下，批次間隔的值會極大地影響系統的數據處理能力。例如，在WordCountNetwork示例 中，對於特定的數據速率，一個系統可能可以在批次間隔爲2秒時跟上數據接收速度，但若是把批次間隔改成500毫秒系統可能就處理不過來了。因此，批次間隔 須要謹慎設置，以確保生產系統可以處理得過來。

要找出適合的批次間隔，你能夠從一個比較保守的批次間隔值（如5~10秒）開始測試。要驗證系統是否能跟上當前的數據接收速率，你可能須要檢查一下端到端的批次處理延遲（能夠看看Spark驅動器log4j日誌中的Total delay，也能夠用StreamingListener接 口來檢測）。若是這個延遲能保持和批次間隔差很少，那麼系統基本就是穩定的。不然，若是這個延遲持久在增加，也就是說系統跟不上數據接收速度，那也就意味 着系統不穩定。一旦系統文檔下來後，你就能夠嘗試提升數據接收速度，或者減小批次間隔值。不過須要注意，瞬間的延遲增加能夠只是暫時的，只要這個延遲後續 會自動降下來就沒有問題（如：降到小於批次間隔值）

---

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.

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.

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.

內存調優

Spark應用內存佔用和GC調優已經在調優指南（Tuning Guide）中有詳細的討論。牆裂建議你讀一讀那篇文檔。本節中，咱們只是討論一下幾個專門用於Spark Streaming的調優參數。

Spark Streaming應用在集羣中佔用的內存量嚴重依賴於具體所使用的tranformation算子。例如，若是想要用一個窗口算子操縱最近10分鐘的數 據，那麼你的集羣至少須要在內存裏保留10分鐘的數據；另外一個例子是updateStateByKey，若是key不少的話，相對應的保存的key的 state也會不少，而這些都須要佔用內存。而若是你的應用只是作一個簡單的 「映射-過濾-存儲」（map-filter-store）操做的話，那須要的內存就不多了。

通常狀況下，streaming接收器接收到的數據會以 StorageLevel.MEMORY_AND_DISK_SER_2 這個存儲級別存到spark中，也就是說，若是內存裝不下，數據將被吐到磁盤上。數據吐到磁盤上會大大下降streaming應用的性能，所以仍是建議根 據你的應用處理的數據量，提供充足的內存。最好就是，一邊小規模地放大內存，再觀察評估，而後再放大，再評估。

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.

There are a few parameters that can help you tune the memory usage and GC overheads:

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

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

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

• Other tips: To further reduce GC overheads, here are some more tips to try.

  • Persist RDDs using the OFF_HEAP storage level. See more detail in the Spark Programming Guide.
  • Use more executors with smaller heap sizes. This will reduce the GC pressure within each JVM heap.

另外一個內存調優的方向就是垃圾回收。由於streaming應用每每都須要低延遲，因此確定不但願出現大量的或耗時較長的JVM垃圾回收暫停。

如下是一些可以幫助你減小內存佔用和GC開銷的參數或手段：

• DStream持久化級別（Persistence Level of DStreams）: 前面數據序列化（Data Serialization） 這小節已經提到過，默認streaming的輸入RDD會被持久化成序列化的字節流。相對於非序列化數據，這樣能夠減小內存佔用和GC開銷。若是啓用 Kryo序列化，還能進一步減小序列化數據大小和內存佔用量。若是你還須要進一步減小內存佔用的話，能夠開啓數據壓縮（經過 spark.rdd.compress這個配置設定），只不過數據壓縮會增長CPU消耗。
• 清除老數據（Clearing old data）: 默認狀況下，全部的輸入數據以及DStream的transformation算子產生的持久化RDD都是自動清理的。Spark Streaming會根據所使用的transformation算子來清理老數據。例如，你用了一個窗口操做處理最近10分鐘的數據，那麼Spark Streaming會保留至少10分鐘的數據，而且會主動把更早的數據都刪掉。固然，你能夠設置 streamingContext.remember 以保留更長時間段的數據（好比：你可能會須要交互式地查詢更老的數據）。
• CMS垃圾回收器（CMS Garbage Collector）: 爲了儘可能減小GC暫停的時間，咱們牆裂建議使用CMS垃圾回收器（concurrent mark-and-sweep GC）。雖然CMS GC會稍微下降系統的整體吞吐量，但咱們仍建議使用它，由於CMS GC能使批次處理的時間保持在一個比較恆定的水平上。最後，你須要確保在驅動器（經過spark-submit中的–driver-java- options設置）和執行器（使用spark.executor.extraJavaOptions配置參數）上都設置了CMS GC。
• 其餘提示: 若是還想進一步減小GC開銷，如下是更進一步的能夠嘗試的手段：
  • 配合Tachyon使用堆外內存來持久化RDD。詳見Spark編程指南（Spark Programming Guide）
  • 使用更多可是更小的執行器進程。這樣GC壓力就會分散到更多的JVM堆中。

---

---

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.

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

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:

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.

With this basic knowledge, let us understand the fault-tolerance semantics of Spark Streaming.

背景

要理解Spark Streaming所提供的容錯語義，咱們首先須要回憶一下Spark RDD所提供的基本容錯語義。

1. RDD是不可變的，可重算的，分佈式數據集。每一個RDD都記錄了其建立算子的血統信息，其中每一個算子都以可容錯的數據集做爲輸入數據。
2. 若是RDD的某個分區由於節點失效而丟失，則該分區能夠根據RDD的血統信息以及相應的原始輸入數據集從新計算出來。
3. 假定全部RDD transformation算子計算過程都是肯定性的，那麼經過這些算子獲得的最終RDD老是包含相同的數據，而與Spark集羣的是否故障無關。

Spark主要操做一些可容錯文件系統的數據，如：HDFS或S3。所以，全部從這些可容錯數據源產生的RDD也是可容錯的。然而，對於Spark Streaming並不是如此，由於多數狀況下Streaming須要從網絡遠端接收數據，這回致使Streaming的數據源並不可靠（尤爲是對於使用了 fileStream的應用）。要實現RDD相同的容錯屬性，數據接收就必須用多個不一樣worker節點上的Spark執行器來實現（默認副本因子是 2）。所以一旦出現故障，系統須要恢復兩種數據：

1. 接收並保存了副本的數據 – 數據不會由於單個worker節點故障而丟失，由於有副本！
2. 接收但還沒有保存副本數據 – 由於數據並無副本，因此一旦故障，只能從數據源從新獲取。

此外，還有兩種可能的故障類型須要考慮：

1. Worker節點故障 – 任何運行執行器的worker節點一旦故障，節點上內存中的數據都會丟失。若是這些節點上有接收器在運行，那麼其包含的緩存數據也會丟失。
2. Driver節點故障 – 若是Spark Streaming的驅動節點故障，那麼很顯然SparkContext對象就沒了，全部執行器及其內存數據也會丟失。

有了以上這些基本知識，下面咱們就進一步瞭解一下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.

定義

流式系統的可靠度語義能夠據此來分類：單條記錄在系統中被處理的次數保證。一個流式系統可能提供保證一定是如下三種之一（無論系統是否出現故障）：

1. 至多一次（At most once）: 每條記錄要麼被處理一次，要麼就沒有處理。
2. 至少一次（At least once）: 每條記錄至少被處理過一次（一次或屢次）。這種保證能確保沒有數據丟失，比「至多一次」要強。但有可能出現數據重複。
3. 精確一次（Exactly once）: 每條記錄都精確地只被處理一次 – 也就是說，既沒有數據丟失，也不會出現數據重複。這是三種保證中最強的一種。

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.

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.

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.

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.

基礎語義

任何流式處理系統通常都會包含如下三個數據處理步驟：

1. 數據接收（Receiving the data）: 從數據源拉取數據。
2. 數據轉換（Transforming the data）: 將接收到的數據進行轉換（使用DStream和RDD transformation算子）。
3. 數據推送（Pushing out the data）: 將轉換後最終數據推送到外部文件系統，數據庫或其餘展現系統。

若是Streaming應用須要作到端到端的「精確一次」的保證，那麼就必須在以上三個步驟中各自都保證精確一次：即，每條記錄必須，只接收一次、處理一次、推送一次。下面讓咱們在Spark Streaming的上下文環境中來理解一下這三個步驟的語義：

1. 數據接收: 不一樣數據源提供的保證不一樣，下一節再詳細討論。
2. 數據轉換: 全部的數據都會被「精確一次」處理，這要歸功於RDD提供的保障。即便出現故障，只要數據源還能訪問，最終所轉換獲得的RDD老是包含相同的內容。
3. 數據推送: 輸出操做默認保證「至少一次」的語義，是否能「精確一次」還要看所使用的輸出算子（是否冪等）以及下游系統（是否支持事務）。不過用戶也能夠開發本身的事務機制來實現「精確一次」語義。這個後續會有詳細討論。

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.

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.

接收數據語義

不一樣的輸入源提供不一樣的數據可靠性級別，從「至少一次」到「精確一次」。

從文件接收數據

若是全部的輸入數據都來源於可容錯的文件系統，如HDFS，那麼Spark Streaming就能在任何故障中恢復並處理全部的數據。這種狀況下就能保證精確一次語義，也就是說無論出現什麼故障，全部的數據老是精確地只處理一次，很少也很多。

基於接收器接收數據

對於基於接收器的輸入源，容錯語義將同時依賴於故障場景和接收器類型。前面也已經提到過，spark Streaming主要有兩種類型的接收器：

1. 可靠接收器 – 這類接收器會在數據接收並保存好副本後，向可靠數據源發送確認信息。這類接收器故障時，是不會給緩存的（已接收但還沒有保存副本）數據發送確認信息。所以，一旦接收器重啓，沒有收到確認的數據，會從新從數據源再獲取一遍，因此即便有故障也不會丟數據。
2. 不可靠接收器 – 這類接收器不會發送確認信息，所以一旦worker和driver出現故障，就有可能會丟失數據。

對於不一樣的接收器，咱們能夠得到以下不一樣的語義。若是一個worker節點故障了，對於可靠接收器來書，不會有數據丟失。而對於不可靠接收器，緩存 的（接收但還沒有保存副本）數據可能會丟失。若是driver節點故障了，除了接收到的數據以外，其餘的已經接收且已經保存了內存副本的數據都會丟失，這將 會影響有狀態算子的計算結果。

爲了不丟失已經收到且保存副本的數，從 spark 1.2 開始引入了WAL（write ahead logs），以便將這些數據寫入到可容錯的存儲中。只要你使用可靠接收器，同時啓用WAL（write ahead logs enabled），那麼久不再用爲數據丟失而擔憂了。而且這時候，還能提供「至少一次」的語義保證。

The following table summarizes the semantics under failures:

Deployment Scenario	Worker Failure	Driver Failure
Spark 1.1 or earlier, OR Spark 1.2 or later without write ahead logs	Buffered data lost with unreliable receivers Zero data loss with reliable receivers At-least once semantics	Buffered data lost with unreliable receivers Past data lost with all receivers Undefined semantics
Spark 1.2 or later with write ahead logs	Zero data loss with reliable receivers At-least once semantics	Zero data loss with reliable receivers and files At-least once semantics

下表總結了故障狀況下的各類語義：

部署場景	Worker 故障	Driver 故障
Spark 1.1及之前版本 或者 Spark 1.2及之後版本，且未開啓WAL	若使用不可靠接收器，則可能丟失緩存（已接收但還沒有保存副本）數據； 若使用可靠接收器，則沒有數據丟失，且提供至少一次處理語義	若使用不可靠接收器，則緩存數據和已保存數據均可能丟失； 若使用可靠接收器，則沒有緩存數據丟失，但已保存數據可能丟失，且不提供語義保證
Spark 1.2及之後版本，並啓用WAL	若使用可靠接收器，則沒有數據丟失，且提供至少一次語義保證	若使用可靠接收器和文件，則無數據丟失，且提供至少一次語義保證

With 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 (experimental as of Spark 2.0.0) is further discussed in the Kafka Integration Guide.

從Kafka Direct API接收數據

從Spark 1.3開始，咱們引入Kafka Direct API，該API能爲Kafka數據源提供「精確一次」語義保證。有了這個輸入API，再加上輸出算子的「精確一次」保證，你就能真正實現端到端的「精確 一次」語義保證。（改功能截止Spark 1.6.1仍是實驗性的）更詳細的說明見：Kafka Integration Guide。

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.

• Idempotent updates: Multiple attempts always write the same data. For example, saveAs***Files always writes the same data to the generated 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.

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

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

 

 

從Kafka Direct API接收數據

從Spark 1.3開始，咱們引入Kafka Direct API，該API能爲Kafka數據源提供「精確一次」語義保證。有了這個輸入API，再加上輸出算子的「精確一次」保證，你就能真正實現端到端的「精確 一次」語義保證。（改功能截止Spark 1.6.1仍是實驗性的）更詳細的說明見：Kafka Integration Guide。

輸出算子的語義

輸出算子（如 foreachRDD）提供「至少一次」語義保證，也就是說，若是worker故障，單條輸出數據可能會被屢次寫入外部實體中。不過這對於文件系統來講是 能夠接受的（使用saveAs***Files 屢次保存文件會覆蓋以前的），因此咱們須要一些額外的工做來實現「精確一次」語義。主要有兩種實現方式：

• 冪等更新（Idempotent updates）: 就是說屢次操做，產生的結果相同。例如，屢次調用saveAs***Files保存的文件老是包含相同的數據。
• 事務更新（Transactional updates）: 全部的更新都是事務性的，這樣一來就能保證更新的原子性。如下是一種實現方式：
  • 用批次時間（在foreachRDD中可用）和分區索引建立一個惟一標識，該標識表明流式應用中惟一的一個數據塊。
  • 基於這個標識創建更新事務，並使用數據塊數據更新外部系統。也就是說，若是該標識未被提交，則原子地將標識表明的數據更新到外部系統。不然，就認爲該標識已經被提交，直接忽略之。

    dstream.foreachRDD { (rdd, time) =>
      rdd.foreachPartition { partitionIterator =>
        val partitionId = TaskContext.get.partitionId()
        val uniqueId = generateUniqueId(time.milliseconds, partitionId)
        // 使用uniqueId做爲事務的惟一標識，基於uniqueId實現partitionIterator所指向數據的原子事務提交
      }
    }

---

---

---

Migration Guide from 0.9.1 or below to 1.x

Between Spark 0.9.1 and Spark 1.0, there were a few API changes made to ensure future API stability. This section elaborates the steps required to migrate your existing code to 1.0.

Input DStreams: All operations that create an input stream (e.g., StreamingContext.socketStream, FlumeUtils.createStream, etc.) now returns InputDStream / ReceiverInputDStream (instead of DStream) for Scala, and JavaInputDStream / JavaPairInputDStream / JavaReceiverInputDStream / JavaPairReceiverInputDStream (instead of JavaDStream) for Java. This ensures that functionality specific to input streams can be added to these classes in the future without breaking binary compatibility. Note that your existing Spark Streaming applications should not require any change (as these new classes are subclasses of DStream/JavaDStream) but may require recompilation with Spark 1.0.

Custom Network Receivers: Since the release to Spark Streaming, custom network receivers could be defined in Scala using the class NetworkReceiver. However, the API was limited in terms of error handling and reporting, and could not be used from Java. Starting Spark 1.0, this class has been replaced by Receiver which has the following advantages.

• Methods like stop and restart have been added to for better control of the lifecycle of a receiver. See the custom receiver guide for more details.
• Custom receivers can be implemented using both Scala and Java.

在Spark 0.9.1和Spark 1.0之間，有一些API接口變動，變動目的是爲了保障將來版本API的穩定。本節將詳細說明一下從已有版本遷移升級到1.0所需的工做。

輸入DStream（Input DStreams）: 全部建立輸入流的算子（如：StreamingContext.socketStream, FlumeUtils.createStream 等）的返回值再也不是DStream（對Java來講是JavaDStream），而是 InputDStream / ReceiverInputDStream（對Java來講是JavaInputDStream / JavaPairInputDStream /JavaReceiverInputDStream / JavaPairReceiverInputDStream）。這樣才能確保特定輸入流的功能可以在將來持續增長到這些class中，而不會打破二進制兼容性。注意，已有的Spark Streaming應用應該不須要任何代碼修改（新的返回類型都是DStream的子類），只不過須要基於Spark 1.0從新編譯一把。

定製網絡接收器（Custom Network Receivers）: 自從Spark Streaming發佈以來，Scala就能基於NetworkReceiver來定製網絡接收器。但因爲錯誤處理和彙報API方便的限制，該類型不能在Java中使用。因此Spark 1.0開始，用 Receiver 來替換掉這個NetworkReceiver，主要的好處以下：

• 該類型新增了stop和restart方法，便於控制接收器的生命週期。詳見custom receiver guide。
• 定製接收器用Scala和Java都能實現。

To migrate your existing custom receivers from the earlier NetworkReceiver to the new Receiver, you have to do the following.

• Make your custom receiver class extend org.apache.spark.streaming.receiver.Receiver instead of org.apache.spark.streaming.dstream.NetworkReceiver.
• Earlier, a BlockGenerator object had to be created by the custom receiver, to which received data was added for being stored in Spark. It had to be explicitly started and stopped from onStart() and onStop() methods. The new Receiver class makes this unnecessary as it adds a set of methods named store(<data>) that can be called to store the data in Spark. So, to migrate your custom network receiver, remove any BlockGenerator object (does not exist any more in Spark 1.0 anyway), and use store(...) methods on received data.

爲了將已有的基於NetworkReceiver的自定義接收器遷移到Receiver上來，你須要以下工做：

• 首先你的自定義接收器類型須要從 org.apache.spark.streaming.receiver.Receiver繼承，而再也不是org.apache.spark.streaming.dstream.NetworkReceiver。
• 原先，咱們須要在自定義接收器中建立一個BlockGenerator來保存接收到的數據。你必須顯示的實現onStart() 和 onStop() 方法。而在新的Receiver class中，這些都不須要了，你只須要調用它的store系列的方法就能將數據保存到Spark中。因此你接下來須要作的遷移工做就是，刪除 BlockGenerator對象（這個類型在Spark 1.0以後也沒有了~），而後用store(…)方法來保存接收到的數據。

Actor-based Receivers: The Actor-based Receiver APIs have been moved to DStream Akka. Please refer to the project for more details.

基於Actor的接收器（Actor-based Receivers）: 從actor class繼承後，並實現了org.apache.spark.streaming.receiver.Receiver 後，便可從Akka Actors中獲取數據。獲取數據的類被重命名爲  org.apache.spark.streaming.receiver.ActorHelper ，而保存數據的pushBlocks(…)方法也被重命名爲 store(…)。其餘org.apache.spark.streaming.receivers包中的工具類也被移到  org.apache.spark.streaming.receiver 包下並重命名，新的類名應該比以前更加清晰。

---

---

Where to Go from Here

更多的參考資料

• Additional guides
  • Kafka Integration Guide
  • Kinesis Integration Guide
  • Custom Receiver Guide
• External DStream data sources:
  • DStream MQTT
  • DStream Twitter
  • DStream Akka
  • DStream ZeroMQ
• API documentation
  • Scala docs
    • StreamingContext and DStream
    • KafkaUtils, FlumeUtils, KinesisUtils,
  • Java docs
    • JavaStreamingContext, JavaDStream and JavaPairDStream
    • KafkaUtils, FlumeUtils, KinesisUtils
  • Python docs
    • StreamingContext and DStream
    • KafkaUtils
• More examples in Scala and Java and Python
• Paper and video describing Spark Streaming.