討論QQ：1586558083

目錄

• 一、簡介
  • 1.1　概述
  • 1.2　一個小栗子
  • 2.2　初始化StreamingContext
  • 2.3　離散數據流 (DStreams)
  • 2.4　輸入DStream和接收器
  • 2.5　接收器可靠性
• 二、基本概念
  • 2.1　鏈接依賴項
• 三、DStream支持的transformation算子
  • 3.1　updateStateByKey算子
  • 3.2　transform算子
  • 3.3　基于窗口（window）的算子
  • 3.4　Join相關算子
• 四、DStream輸出算子
  • 4.1　使用foreachRDD的設計模式

?

正文

官網地址：http://spark.apache.org/docs/latest/streaming-programming-guide.html

回到頂部

一、簡介

1.1　概述

Spark Streaming?是Spark核心API的一個擴展，可以實現高吞吐量的、具備容錯機制的實時流數據的處理。支持從多種數據源獲取數據，包括Kafk、Flume、Twitter、ZeroMQ、Kinesis?以及TCP sockets，從數據源獲取數據之后，可以使用諸如map、reduce、join和window等高級函數進行復雜算法的處理。最后還可以將處理結果存儲到文件系統，數據庫和現場儀表盤。在“One Stack rule them all”的基礎上，還可以使用Spark的其他子框架，如機器學習、圖計算等，對流數據進行處理。

Spark Streaming處理的數據流圖：

Spark的各個子框架，都是基于核心Spark的，Spark Streaming在內部的處理機制是，接收實時流的數據，并根據一定的時間間隔拆分成一批批的數據，然后通過Spark Engine處理這些批數據，最終得到處理后的一批批結果數據。

Spark Streaming為這種持續的數據流提供了的一個高級抽象，即：discretized stream（離散數據流）或者叫DStream。DStream既可以從輸入數據源創建得來，如：Kafka、Flume或者Kinesis，也可以從其他DStream經一些算子操作得到。其實在內部，一個DStream就是包含了一系列RDDs。

本文檔將向你展示如何用DStream進行Spark Streaming編程。Spark Streaming支持Scala、Java和Python（始于Spark 1.2），本文檔的示例包括這三種語言。

注意：對Python來說，有一部分API尚不支持，或者是和Scala、Java不同。本文檔中會用高亮形式來注明這部分?Python API。

1.2　一個小栗子

在深入Spark Streaming編程細節之前，我們先來看看一個簡單的小栗子以便有個感性認識。假設我們在一個TCP端口上監聽一個數據服務器的數據，并對收到的文本數據中的單詞計數。以下你所需的全部工作：

• Scala
• Java
• Python

首先，我們需要導入Spark Streaming的相關class的一些包，以及一些支持StreamingContext隱式轉換的包（這些隱式轉換能給DStream之類的class增加一些有用的方法）。StreamingContext?是Spark Streaming的入口。我們將會創建一個本地 StreamingContext對象，包含兩個執行線程，并將批次間隔設為1秒。

import org.apache.spark._ import org.apache.spark.streaming._ import org.apache.spark.streaming.StreamingContext._ // 從Spark 1.3之后這行就可以不需要了// 創建一個local StreamingContext，包含2個工作線程，并將批次間隔設為1秒 // master至少需要2個CPU核，以避免出現任務餓死的情況 val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount") val ssc = new StreamingContext(conf, Seconds(1))

利用這個上下文對象（StreamingContext），我們可以創建一個DStream，該DStream代表從前面的TCP數據源流入的數據流，同時TCP數據源是由主機名（如：hostnam）和端口（如：9999）來描述的。

// 創建一個連接到hostname:port的DStream，如：localhost:9999 val lines = ssc.socketTextStream("localhost", 9999)

這里的 lines 就是從數據server接收到的數據流。其中每一條記錄都是一行文本。接下來，我們就需要把這些文本行按空格分割成單詞。

// 將每一行分割成多個單詞 val words = lines.flatMap(_.split(" "))

flatMap 是一種 “一到多”（one-to-many）的映射算子，它可以將源DStream中每一條記錄映射成多條記錄，從而產生一個新的DStream對象。在本例中，lines中的每一行都會被flatMap映射為多個單詞，從而生成新的words DStream對象。然后，我們就能對這些單詞進行計數了。

import org.apache.spark.streaming.StreamingContext._ // Spark 1.3之后不再需要這行 // 對每一批次中的單詞進行計數 val pairs = words.map(word => (word, 1)) val wordCounts = pairs.reduceByKey(_ + _) // 將該DStream產生的RDD的頭十個元素打印到控制臺上 wordCounts.print()

words這個DStream對象經過map算子（一到一的映射）轉換為一個包含（word, 1）鍵值對的DStream對象pairs，再對pairs使用reduce算子，得到每個批次中各個單詞的出現頻率。最后，wordCounts.print() 將會每秒（前面設定的批次間隔）打印一些單詞計數到控制臺上。

注意，執行以上代碼后，Spark Streaming只是將計算邏輯設置好，此時并未真正的開始處理數據。要啟動之前的處理邏輯，我們還需要如下調用：

ssc.start() // 啟動流式計算 ssc.awaitTermination() // 等待直到計算終止

完整的代碼可以在Spark Streaming的例子?NetworkWordCount?中找到。

如果你已經有一個Spark包（下載在這里downloaded，自定義構建在這里built），就可以執行按如下步驟運行這個例子。

首先，你需要運行netcat（Unix-like系統都會有這個小工具），將其作為data server

$ nc -lk 9999

然后，在另一個終端，按如下指令執行這個例子

• Scala
• Java
• Python

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

好了，現在你嘗試可以在運行netcat的終端里敲幾個單詞，你會發現這些單詞以及相應的計數會出現在啟動Spark Streaming例子的終端屏幕上。看上去應該和下面這個示意圖類似：

回到頂部

二、基本概念

下面，我們在之前的小栗子基礎上，繼續深入了解一下Spark Streaming的一些基本概念。

2.1　鏈接依賴項

和Spark類似，Spark Streaming也能在Maven庫中找到。如果你需要編寫Spark Streaming程序，你就需要將以下依賴加入到你的SBT或Maven工程依賴中。

• Maven
• SBT

<dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.10</artifactId><version>1.6.1</version> </dependency>

還有，對于從Kafka、Flume以及Kinesis這類數據源提取數據的流式應用來說，還需要額外增加相應的依賴項，下表列出了各種數據源對應的額外依賴項：

數據源Maven工件

Kafka	spark-streaming-kafka_2.10
Flume	spark-streaming-flume_2.10
Kinesis	spark-streaming-kinesis-asl_2.10 [Amazon Software License]
Twitter	spark-streaming-twitter_2.10
ZeroMQ	spark-streaming-zeromq_2.10
MQTT	spark-streaming-mqtt_2.10

最新的依賴項信息（包括源代碼和Maven工件）請參考Maven repository。

2.2　初始化StreamingContext

要初始化任何一個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))

上面代碼中的 appName 是你給該應用起的名字，這個名字會展示在Spark集群的web UI上。而 master 是Spark, Mesos or YARN cluster URL，如果支持本地測試，你也可以用”local[*]”為其賦值。通常在實際工作中，你不應該將master參數硬編碼到代碼里，而是應用通過spark-submit的參數來傳遞master的值（launch the application with?spark-submit?）。不過對本地測試來說，”local[*]”足夠了（該值傳給master后，Spark Streaming將在本地進程中，啟動n個線程運行，n與本地系統CPU core數相同）。注意，StreamingContext在內部會創建一個??SparkContext?對象（SparkContext是所有Spark應用的入口，在StreamingContext對象中可以這樣訪問：ssc.sparkContext）。

StreamingContext還有另一個構造參數，即：批次間隔，這個值的大小需要根據應用的具體需求和可用的集群資源來確定。詳見Spark性能調優（?Performance Tuning）。

StreamingContext對象也可以通過已有的SparkContext對象來創建，示例如下：

import org.apache.spark.streaming._val sc = ... // 已有的SparkContext val ssc = new StreamingContext(sc, Seconds(1))

context對象創建后，你還需要如下步驟：

創建DStream對象，并定義好輸入數據源。

基于數據源DStream定義好計算邏輯和輸出。

調用streamingContext.start() 啟動接收并處理數據。

調用streamingContext.awaitTermination() 等待流式處理結束（不管是手動結束，還是發生異常錯誤）

你可以主動調用 streamingContext.stop() 來手動停止處理流程。

需要關注的重點:

• 一旦streamingContext啟動，就不能再對其計算邏輯進行添加或修改。
• 一旦streamingContext被stop掉，就不能restart。
• 單個JVM虛機同一時間只能包含一個active的StreamingContext。
• StreamingContext.stop() 也會把關聯的SparkContext對象stop掉，如果不想把SparkContext對象也stop掉，可以將StreamingContext.stop的可選參數 stopSparkContext 設為false。
• 一個SparkContext對象可以和多個StreamingContext對象關聯，只要先對前一個StreamingContext.stop(sparkContext=false)，然后再創建新的StreamingContext對象即可。

2.3　離散數據流 (DStreams)

離散數據流（DStream）是Spark Streaming最基本的抽象。它代表了一種連續的數據流，要么從某種數據源提取數據，要么從其他數據流映射轉換而來。DStream內部是由一系列連續的RDD組成的，每個RDD都是不可變、分布式的數據集（詳見Spark編程指南 –?Spark Programming Guide）。每個RDD都包含了特定時間間隔內的一批數據，如下圖所示：

任何作用于DStream的算子，其實都會被轉化為對其內部RDD的操作。例如，在前面的例子中，我們將 lines 這個DStream轉成words DStream對象，其實作用于lines上的flatMap算子，會施加于lines中的每個RDD上，并生成新的對應的RDD，而這些新生成的RDD對象就組成了words這個DStream對象。其過程如下圖所示：

底層的RDD轉換仍然是由Spark引擎來計算。DStream的算子將這些細節隱藏了起來，并為開發者提供了更為方便的高級API。后續會詳細討論這些高級算子。

2.4　輸入DStream和接收器

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

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

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

本節中，我們將會從每種數據源中挑幾個繼續深入討論。

注意，如果你需要同時從多個數據源拉取數據，那么你就需要創建多個DStream對象（詳見后續的性能調優這一小節）。多個DStream對象其實也就同時創建了多個數據流接收器。但是請注意，Spark的worker/executor 都是長期運行的，因此它們都會各自占用一個分配給Spark Streaming應用的CPU。所以，在運行Spark Streaming應用的時候，需要注意分配足夠的CPU core（本地運行時，需要足夠的線程）來處理接收到的數據，同時還要足夠的CPU core來運行這些接收器。

要點

• 如果本地運行Spark Streaming應用，記得不能將master設為”local” 或 “local[1]”。這兩個值都只會在本地啟動一個線程。而如果此時你使用一個包含接收器（如：套接字、Kafka、Flume等）的輸入DStream，那么這一個線程只能用于運行這個接收器，而處理數據的邏輯就沒有線程來執行了。因此，本地運行時，一定要將master設為”local[n]”，其中 n > 接收器的個數（有關master的詳情請參考Spark Properties）。
• 將Spark Streaming應用置于集群中運行時，同樣，分配給該應用的CPU core數必須大于接收器的總數。否則，該應用就只會接收數據，而不會處理數據。

基礎數據源

前面的小栗子中，我們已經看到，使用ssc.socketTextStream(…) 可以從一個TCP連接中接收文本數據。而除了TCP套接字外，StreamingContext API 還支持從文件或者Akka actor中拉取數據。

文件數據流（File Streams）:?可以從任何兼容HDFS API（包括：HDFS、S3、NFS等）的文件系統，創建方式如下：

• • Scala
  • Java
  • Python

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

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

• 各個文件數據格式必須一致。
• dataDirectory中的文件必須通過moving或者renaming來創建。
• 一旦文件move進dataDirectory之后，就不能再改動。所以如果這個文件后續還有寫入，這些新寫入的數據不會被讀取。

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

另外，文件數據流不是基于接收器的，所以不需要為其單獨分配一個CPU core。

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

• 基于自定義Actor的數據流（Streams based on Custom Actors）:?DStream可以由Akka actor創建得到，只需調用 streamingContext.actorStream(actorProps, actor-name)。詳見自定義接收器（Custom Receiver Guide）。actorStream暫時不支持Python API。
• RDD隊列數據流（Queue of RDDs as a Stream）:?如果需要測試Spark Streaming應用，你可以創建一個基于一批RDD的DStream對象，只需調用 streamingContext.queueStream(queueOfRDDs)。RDD會被一個個依次推入隊列，而DStream則會依次以數據流形式處理這些RDD的數據。

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

高級數據源

Python API?自 Spark 1.6.1 起，Kafka、Kinesis、Flume和MQTT這些數據源將支持Python。

使用這類數據源需要依賴一些額外的代碼庫，有些依賴還挺復雜的（如：Kafka、Flume）。因此為了減少依賴項版本沖突問題，各個數據源DStream的相關功能被分割到不同的代碼包中，只有用到的時候才需要鏈接打包進來。例如，如果你需要使用Twitter的tweets作為數據源，你需要以下步驟：

Linking: 將spark-streaming-twitter_2.10工件加入到SBT/Maven項目依賴中。

Programming: 導入TwitterUtils class，然后調用 TwitterUtils.createStream 創建一個DStream，具體代碼見下放。

Deploying: 生成一個uber Jar包，并包含其所有依賴項（包括 spark-streaming-twitter_2.10及其自身的依賴樹），再部署這個Jar包。部署詳情請參考部署這一節（Deploying section）。

• Scala
• Java

import org.apache.spark.streaming.twitter._TwitterUtils.createStream(ssc, None)

注意，高級數據源在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。

自定義數據源

Python API?自定義數據源目前還不支持Python。

輸入DStream也可以用自定義的方式創建。你需要做的只是實現一個自定義的接收器（receiver），以便從自定義的數據源接收數據，然后將數據推入Spark中。詳情請參考自定義接收器指南（Custom Receiver Guide）。

2.5　接收器可靠性

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

可靠接收器（Reliable Receiver）?– 可靠接收器會在成功接收并保存好Spark數據副本后，向可靠數據源發送確認信息。

不可靠接收器（Unreliable Receiver）?– 不可靠接收器不會發送任何確認信息。不過這種接收器常用語于不支持確認的數據源，或者不想引入數據確認的復雜性的數據源。

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

---

回到頂部

三、DStream支持的transformation算子

和RDD類似，DStream也支持從輸入DStream經過各種transformation算子映射成新的DStream。DStream支持很多RDD上常見的transformation算子，一些常用的見下表：

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算子深入討論一下。

3.1　updateStateByKey算子

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

定義狀態 – 狀態數據可以是任意類型。

定義狀態更新函數 – 定義好一個函數，其輸入為數據流之前的狀態和新的數據流數據，且可其更新步驟1中定義的輸入數據流的狀態。

在每一個批次數據到達后，Spark都會調用狀態更新函數，來更新所有已有key（不管key是否存在于本批次中）的狀態。如果狀態更新函數返回None，則對應的鍵值對會被刪除。

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

• Scala
• Java
• Python

def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {val newCount = ... // 將新的計數值和之前的狀態值相加，得到新的計數值 Some(newCount) }

該狀態更新函數可以作用于一個包括(word, 1) 鍵值對的DStream上（見本文開頭的小栗子）。

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

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

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

3.2　transform算子

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 val cleanedDStream = wordCounts.transform(rdd => {rdd.join(spamInfoRDD).filter(...) // 將DStream中的RDD和spamInfoRDD關聯，并實時過濾垃圾數據 ... })

注意，這里transform包含的算子，其調用時間間隔和批次間隔是相同的。所以你可以基于時間改變對RDD的操作，如：在不同批次，調用不同的RDD算子，設置不同的RDD分區或者廣播變量等。

3.3　基于窗口（window）的算子

Spark Streaming同樣也提供基于時間窗口的計算，也就是說，你可以對某一個滑動時間窗內的數據施加特定tranformation算子。如下圖所示：

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

• （窗口長度）window length?– 窗口覆蓋的時間長度（上圖中為3）
• （滑動距離）sliding interval?– 窗口啟動的時間間隔（上圖中為2）

注意，這兩個參數都必須是DStream批次間隔（上圖中為1）的整數倍.

下面咱們舉個栗子。假設，你需要擴展前面的那個小栗子，你需要每隔10秒統計一下前30秒內的單詞計數。為此，我們需要在包含(word, 1)鍵值對的DStream上，對最近30秒的數據調用reduceByKey算子。不過這些都可以簡單地用一個 reduceByKeyAndWindow搞定。

• Scala
• Java
• Python

// 每隔10秒歸約一次最近30秒的數據 val windowedWordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(30), Seconds(10))

?

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

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來指定并行任務數。
?	?

3.4　Join相關算子

最后，值得一提的是，你在Spark Streaming中做各種關聯（join）操作非常簡單。

流-流（Stream-stream）關聯

一個數據流可以和另一個數據流直接關聯。

• • Scala
  • Java
  • Python

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

上面代碼中，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）關聯

其實這種情況已經在前面的DStream.transform算子中介紹過了，這里再舉個基于滑動窗口的例子。

• Scala
• Java
• Python

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

實際上，在上面代碼里，你可以動態地該表join的數據集（dataset）。傳給tranform算子的操作函數會在每個批次重新求值，所以每次該函數都會用最新的dataset值，所以不同批次間你可以改變dataset的值。

完整的DStream transformation算子列表見API文檔。Scala請參考?DStream?和?PairDStreamFunctions. Java請參考?JavaDStream?和?JavaPairDStream. Python見?DStream。

回到頂部

四、DStream輸出算子

輸出算子可以將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的實際計算過程。
?	?

4.1　使用foreachRDD的設計模式

DStream.foreachRDD是一個非常強大的原生工具函數，用戶可以基于此算子將DStream數據推送到外部系統中。不過用戶需要了解如何正確而高效地使用這個工具。以下列舉了一些常見的錯誤。

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

• • Scala
  • Python

dstream.foreachRDD { rdd =>val connection = createNewConnection() // 這行在驅動器（driver）進程執行rdd.foreach { record =>connection.send(record) // 而這行將在worker節點上執行 } }

這段代碼是錯誤的，因為它需要把連接對象序列化，再從驅動器節點發送到worker節點。而這些連接對象通常都是不能跨節點（機器）傳遞的。比如，連接對象通常都不能序列化，或者在另一個進程中反序列化后再次初始化（連接對象通常都需要初始化，因此從驅動節點發到worker節點后可能需要重新初始化）等。解決此類錯誤的辦法就是在worker節點上創建連接對象。

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

• Scala
• Python

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

一般來說，連接對象是有時間和資源開銷限制的。因此，對每條記錄都進行一次連接對象的創建和銷毀會增加很多不必要的開銷，同時也大大減小了系統的吞吐量。一個比較好的解決方案是使用 rdd.foreachPartition – 為RDD的每個分區創建一個單獨的連接對象，示例如下：

• Scala
• Python

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

?

這樣一來，連接對象的創建開銷就攤到很多條記錄上了。

最后，還有一個更優化的辦法，就是在多個RDD批次之間復用連接對象。開發者可以維護一個靜態連接池來保存連接對象，以便在不同批次的多個RDD之間共享同一組連接對象，示例如下：

• Scala
• Python

dstream.foreachRDD { rdd =>rdd.foreachPartition { partitionOfRecords =>// ConnectionPool 是一個靜態的、懶惰初始化的連接池val connection = ConnectionPool.getConnection()partitionOfRecords.foreach(record => connection.send(record)) ConnectionPool.returnConnection(connection) // 將連接返還給連接池，以便后續復用之 } }

?

注意，連接池中的連接應該是懶惰創建的，并且有確定的超時時間，超時后自動銷毀。這個實現應該是目前發送數據最高效的實現方式。

其他要點:

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

?

?轉自：https://yq.aliyun.com/articles/86239?spm=a2c4e.11153940.blogcont86244.110.4e80c973zadb30

?

轉載于:https://www.cnblogs.com/liuys635/p/11002867.html

總結

以上是生活随笔為你收集整理的Spark学习之路 （二十二）SparkStreaming的官方文档的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。