討論QQ：1586558083

目錄

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

?

正文

官網(wǎng)地址：http://spark.apache.org/docs/latest/streaming-programming-guide.html

回到頂部

一、簡(jiǎn)介

1.1　概述

Spark Streaming?是Spark核心API的一個(gè)擴(kuò)展，可以實(shí)現(xiàn)高吞吐量的、具備容錯(cuò)機(jī)制的實(shí)時(shí)流數(shù)據(jù)的處理。支持從多種數(shù)據(jù)源獲取數(shù)據(jù)，包括Kafk、Flume、Twitter、ZeroMQ、Kinesis?以及TCP sockets，從數(shù)據(jù)源獲取數(shù)據(jù)之后，可以使用諸如map、reduce、join和window等高級(jí)函數(shù)進(jìn)行復(fù)雜算法的處理。最后還可以將處理結(jié)果存儲(chǔ)到文件系統(tǒng)，數(shù)據(jù)庫(kù)和現(xiàn)場(chǎng)儀表盤。在“One Stack rule them all”的基礎(chǔ)上，還可以使用Spark的其他子框架，如機(jī)器學(xué)習(xí)、圖計(jì)算等，對(duì)流數(shù)據(jù)進(jìn)行處理。

Spark Streaming處理的數(shù)據(jù)流圖：

Spark的各個(gè)子框架，都是基于核心Spark的，Spark Streaming在內(nèi)部的處理機(jī)制是，接收實(shí)時(shí)流的數(shù)據(jù)，并根據(jù)一定的時(shí)間間隔拆分成一批批的數(shù)據(jù)，然后通過(guò)Spark Engine處理這些批數(shù)據(jù)，最終得到處理后的一批批結(jié)果數(shù)據(jù)。

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

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

注意：對(duì)Python來(lái)說(shuō)，有一部分API尚不支持，或者是和Scala、Java不同。本文檔中會(huì)用高亮形式來(lái)注明這部分?Python API。

1.2　一個(gè)小栗子

在深入Spark Streaming編程細(xì)節(jié)之前，我們先來(lái)看看一個(gè)簡(jiǎn)單的小栗子以便有個(gè)感性認(rèn)識(shí)。假設(shè)我們?cè)谝粋€(gè)TCP端口上監(jiān)聽一個(gè)數(shù)據(jù)服務(wù)器的數(shù)據(jù)，并對(duì)收到的文本數(shù)據(jù)中的單詞計(jì)數(shù)。以下你所需的全部工作：

• Scala
• Java
• Python

首先，我們需要導(dǎo)入Spark Streaming的相關(guān)class的一些包，以及一些支持StreamingContext隱式轉(zhuǎn)換的包（這些隱式轉(zhuǎn)換能給DStream之類的class增加一些有用的方法）。StreamingContext?是Spark Streaming的入口。我們將會(huì)創(chuàng)建一個(gè)本地 StreamingContext對(duì)象，包含兩個(gè)執(zhí)行線程，并將批次間隔設(shè)為1秒。

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

利用這個(gè)上下文對(duì)象（StreamingContext），我們可以創(chuàng)建一個(gè)DStream，該DStream代表從前面的TCP數(shù)據(jù)源流入的數(shù)據(jù)流，同時(shí)TCP數(shù)據(jù)源是由主機(jī)名（如：hostnam）和端口（如：9999）來(lái)描述的。

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

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

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

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

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

words這個(gè)DStream對(duì)象經(jīng)過(guò)map算子（一到一的映射）轉(zhuǎn)換為一個(gè)包含（word, 1）鍵值對(duì)的DStream對(duì)象pairs，再對(duì)pairs使用reduce算子，得到每個(gè)批次中各個(gè)單詞的出現(xiàn)頻率。最后，wordCounts.print() 將會(huì)每秒（前面設(shè)定的批次間隔）打印一些單詞計(jì)數(shù)到控制臺(tái)上。

注意，執(zhí)行以上代碼后，Spark Streaming只是將計(jì)算邏輯設(shè)置好，此時(shí)并未真正的開始處理數(shù)據(jù)。要啟動(dòng)之前的處理邏輯，我們還需要如下調(diào)用：

ssc.start() // 啟動(dòng)流式計(jì)算 ssc.awaitTermination() // 等待直到計(jì)算終止

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

如果你已經(jīng)有一個(gè)Spark包（下載在這里downloaded，自定義構(gòu)建在這里built），就可以執(zhí)行按如下步驟運(yùn)行這個(gè)例子。

首先，你需要運(yùn)行netcat（Unix-like系統(tǒng)都會(huì)有這個(gè)小工具），將其作為data server

$ nc -lk 9999

然后，在另一個(gè)終端，按如下指令執(zhí)行這個(gè)例子

• Scala
• Java
• Python

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

好了，現(xiàn)在你嘗試可以在運(yùn)行netcat的終端里敲幾個(gè)單詞，你會(huì)發(fā)現(xiàn)這些單詞以及相應(yīng)的計(jì)數(shù)會(huì)出現(xiàn)在啟動(dòng)Spark Streaming例子的終端屏幕上。看上去應(yīng)該和下面這個(gè)示意圖類似：

回到頂部

二、基本概念

下面，我們?cè)谥暗男±踝踊A(chǔ)上，繼續(xù)深入了解一下Spark Streaming的一些基本概念。

2.1　鏈接依賴項(xiàng)

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

• Maven
• SBT

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

還有，對(duì)于從Kafka、Flume以及Kinesis這類數(shù)據(jù)源提取數(shù)據(jù)的流式應(yīng)用來(lái)說(shuō)，還需要額外增加相應(yīng)的依賴項(xiàng)，下表列出了各種數(shù)據(jù)源對(duì)應(yīng)的額外依賴項(xiàng)：

數(shù)據(jù)源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

最新的依賴項(xiàng)信息（包括源代碼和Maven工件）請(qǐng)參考Maven repository。

2.2　初始化StreamingContext

要初始化任何一個(gè)Spark Streaming程序，都需要在入口代碼中創(chuàng)建一個(gè)StreamingContext對(duì)象。

• Scala
• Java
• Python

A?StreamingContext?object can be created from a?SparkConf?object.

而StreamingContext對(duì)象需要一個(gè)SparkConf對(duì)象作為其構(gòu)造參數(shù)。

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 是你給該應(yīng)用起的名字，這個(gè)名字會(huì)展示在Spark集群的web UI上。而 master 是Spark, Mesos or YARN cluster URL，如果支持本地測(cè)試，你也可以用”local[*]”為其賦值。通常在實(shí)際工作中，你不應(yīng)該將master參數(shù)硬編碼到代碼里，而是應(yīng)用通過(guò)spark-submit的參數(shù)來(lái)傳遞master的值（launch the application with?spark-submit?）。不過(guò)對(duì)本地測(cè)試來(lái)說(shuō)，”local[*]”足夠了（該值傳給master后，Spark Streaming將在本地進(jìn)程中，啟動(dòng)n個(gè)線程運(yùn)行，n與本地系統(tǒng)CPU core數(shù)相同）。注意，StreamingContext在內(nèi)部會(huì)創(chuàng)建一個(gè)??SparkContext?對(duì)象（SparkContext是所有Spark應(yīng)用的入口，在StreamingContext對(duì)象中可以這樣訪問(wèn)：ssc.sparkContext）。

StreamingContext還有另一個(gè)構(gòu)造參數(shù)，即：批次間隔，這個(gè)值的大小需要根據(jù)應(yīng)用的具體需求和可用的集群資源來(lái)確定。詳見Spark性能調(diào)優(yōu)（?Performance Tuning）。

StreamingContext對(duì)象也可以通過(guò)已有的SparkContext對(duì)象來(lái)創(chuàng)建，示例如下：

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

context對(duì)象創(chuàng)建后，你還需要如下步驟：

創(chuàng)建DStream對(duì)象，并定義好輸入數(shù)據(jù)源。

基于數(shù)據(jù)源DStream定義好計(jì)算邏輯和輸出。

調(diào)用streamingContext.start() 啟動(dòng)接收并處理數(shù)據(jù)。

調(diào)用streamingContext.awaitTermination() 等待流式處理結(jié)束（不管是手動(dòng)結(jié)束，還是發(fā)生異常錯(cuò)誤）

你可以主動(dòng)調(diào)用 streamingContext.stop() 來(lái)手動(dòng)停止處理流程。

需要關(guān)注的重點(diǎn):

• 一旦streamingContext啟動(dòng)，就不能再對(duì)其計(jì)算邏輯進(jìn)行添加或修改。
• 一旦streamingContext被stop掉，就不能restart。
• 單個(gè)JVM虛機(jī)同一時(shí)間只能包含一個(gè)active的StreamingContext。
• StreamingContext.stop() 也會(huì)把關(guān)聯(lián)的SparkContext對(duì)象stop掉，如果不想把SparkContext對(duì)象也stop掉，可以將StreamingContext.stop的可選參數(shù) stopSparkContext 設(shè)為false。
• 一個(gè)SparkContext對(duì)象可以和多個(gè)StreamingContext對(duì)象關(guān)聯(lián)，只要先對(duì)前一個(gè)StreamingContext.stop(sparkContext=false)，然后再創(chuàng)建新的StreamingContext對(duì)象即可。

2.3　離散數(shù)據(jù)流 (DStreams)

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

任何作用于DStream的算子，其實(shí)都會(huì)被轉(zhuǎn)化為對(duì)其內(nèi)部RDD的操作。例如，在前面的例子中，我們將 lines 這個(gè)DStream轉(zhuǎn)成words DStream對(duì)象，其實(shí)作用于lines上的flatMap算子，會(huì)施加于lines中的每個(gè)RDD上，并生成新的對(duì)應(yīng)的RDD，而這些新生成的RDD對(duì)象就組成了words這個(gè)DStream對(duì)象。其過(guò)程如下圖所示：

底層的RDD轉(zhuǎn)換仍然是由Spark引擎來(lái)計(jì)算。DStream的算子將這些細(xì)節(jié)隱藏了起來(lái)，并為開發(fā)者提供了更為方便的高級(jí)API。后續(xù)會(huì)詳細(xì)討論這些高級(jí)算子。

2.4　輸入DStream和接收器

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

Spark Streaming主要提供兩種內(nèi)建的流式數(shù)據(jù)源：

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

本節(jié)中，我們將會(huì)從每種數(shù)據(jù)源中挑幾個(gè)繼續(xù)深入討論。

注意，如果你需要同時(shí)從多個(gè)數(shù)據(jù)源拉取數(shù)據(jù)，那么你就需要?jiǎng)?chuàng)建多個(gè)DStream對(duì)象（詳見后續(xù)的性能調(diào)優(yōu)這一小節(jié)）。多個(gè)DStream對(duì)象其實(shí)也就同時(shí)創(chuàng)建了多個(gè)數(shù)據(jù)流接收器。但是請(qǐng)注意，Spark的worker/executor 都是長(zhǎng)期運(yùn)行的，因此它們都會(huì)各自占用一個(gè)分配給Spark Streaming應(yīng)用的CPU。所以，在運(yùn)行Spark Streaming應(yīng)用的時(shí)候，需要注意分配足夠的CPU core（本地運(yùn)行時(shí)，需要足夠的線程）來(lái)處理接收到的數(shù)據(jù)，同時(shí)還要足夠的CPU core來(lái)運(yùn)行這些接收器。

要點(diǎn)

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

基礎(chǔ)數(shù)據(jù)源

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

文件數(shù)據(jù)流（File Streams）:?可以從任何兼容HDFS API（包括：HDFS、S3、NFS等）的文件系統(tǒng)，創(chuàng)建方式如下：

• • Scala
  • Java
  • Python

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

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

• 各個(gè)文件數(shù)據(jù)格式必須一致。
• dataDirectory中的文件必須通過(guò)moving或者renaming來(lái)創(chuàng)建。
• 一旦文件move進(jìn)dataDirectory之后，就不能再改動(dòng)。所以如果這個(gè)文件后續(xù)還有寫入，這些新寫入的數(shù)據(jù)不會(huì)被讀取。

對(duì)于簡(jiǎn)單的文本文件，更簡(jiǎn)單的方式是調(diào)用 streamingContext.textFileStream(dataDirectory)。

另外，文件數(shù)據(jù)流不是基于接收器的，所以不需要為其單獨(dú)分配一個(gè)CPU core。

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

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

關(guān)于套接字、文件以及Akka actor數(shù)據(jù)流更詳細(xì)信息，請(qǐng)參考相關(guān)文檔：StreamingContext?for Scala,JavaStreamingContext?for Java, and?StreamingContext?for Python。

高級(jí)數(shù)據(jù)源

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

使用這類數(shù)據(jù)源需要依賴一些額外的代碼庫(kù)，有些依賴還挺復(fù)雜的（如：Kafka、Flume）。因此為了減少依賴項(xiàng)版本沖突問(wèn)題，各個(gè)數(shù)據(jù)源DStream的相關(guān)功能被分割到不同的代碼包中，只有用到的時(shí)候才需要鏈接打包進(jìn)來(lái)。例如，如果你需要使用Twitter的tweets作為數(shù)據(jù)源，你需要以下步驟：

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

Programming: 導(dǎo)入TwitterUtils class，然后調(diào)用 TwitterUtils.createStream 創(chuàng)建一個(gè)DStream，具體代碼見下放。

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

• Scala
• Java

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

注意，高級(jí)數(shù)據(jù)源在spark-shell中不可用，因此不能用spark-shell來(lái)測(cè)試基于高級(jí)數(shù)據(jù)源的應(yīng)用。如果真有需要的話，你需要自行下載相應(yīng)數(shù)據(jù)源的Maven工件及其依賴項(xiàng)，并將這些Jar包部署到spark-shell的classpath中。

下面列舉了一些高級(jí)數(shù)據(jù)源：

• 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 通過(guò)?Twitter’s Streaming API?拉取公開tweets數(shù)據(jù)流。認(rèn)證信息可以用任何Twitter4j所支持的方法（methods）。你可以獲取所有的公開數(shù)據(jù)流，當(dāng)然也可以基于某些關(guān)鍵詞進(jìn)行過(guò)濾。示例可以參考TwitterPopularTags?和?TwitterAlgebirdCMS。

自定義數(shù)據(jù)源

Python API?自定義數(shù)據(jù)源目前還不支持Python。

輸入DStream也可以用自定義的方式創(chuàng)建。你需要做的只是實(shí)現(xiàn)一個(gè)自定義的接收器（receiver），以便從自定義的數(shù)據(jù)源接收數(shù)據(jù)，然后將數(shù)據(jù)推入Spark中。詳情請(qǐng)參考自定義接收器指南（Custom Receiver Guide）。

2.5　接收器可靠性

從可靠性角度來(lái)劃分，大致有兩種數(shù)據(jù)源。其中，像Kafka、Flume這樣的數(shù)據(jù)源，它們支持對(duì)所傳輸?shù)臄?shù)據(jù)進(jìn)行確認(rèn)。系統(tǒng)收到這類可靠數(shù)據(jù)源過(guò)來(lái)的數(shù)據(jù)，然后發(fā)出確認(rèn)信息，這樣就能夠確保任何失敗情況下，都不會(huì)丟數(shù)據(jù)。因此我們可以將接收器也相應(yīng)地分為兩類：

可靠接收器（Reliable Receiver）?– 可靠接收器會(huì)在成功接收并保存好Spark數(shù)據(jù)副本后，向可靠數(shù)據(jù)源發(fā)送確認(rèn)信息。

不可靠接收器（Unreliable Receiver）?– 不可靠接收器不會(huì)發(fā)送任何確認(rèn)信息。不過(guò)這種接收器常用語(yǔ)于不支持確認(rèn)的數(shù)據(jù)源，或者不想引入數(shù)據(jù)確認(rèn)的復(fù)雜性的數(shù)據(jù)源。

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

---

回到頂部

三、DStream支持的transformation算子

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

Transformation算子用途

map(func)	返回會(huì)一個(gè)新的DStream，并將源DStream中每個(gè)元素通過(guò)func映射為新的元素
flatMap(func)	和map類似，不過(guò)每個(gè)輸入元素不再是映射為一個(gè)輸出，而是映射為0到多個(gè)輸出
filter(func)	返回一個(gè)新的DStream，并包含源DStream中被func選中（func返回true）的元素
repartition(numPartitions)	更改DStream的并行度（增加或減少分區(qū)數(shù)）
union(otherStream)	返回新的DStream，包含源DStream和otherDStream元素的并集
count()	返回一個(gè)包含單元素RDDs的DStream，其中每個(gè)元素是源DStream中各個(gè)RDD中的元素個(gè)數(shù)
reduce(func)	返回一個(gè)包含單元素RDDs的DStream，其中每個(gè)元素是通過(guò)源RDD中各個(gè)RDD的元素經(jīng)func（func輸入兩個(gè)參數(shù)并返回一個(gè)同類型結(jié)果數(shù)據(jù)）聚合得到的結(jié)果。func必須滿足結(jié)合律，以便支持并行計(jì)算。
countByValue()	如果源DStream包含的元素類型為K，那么該算子返回新的DStream包含元素為(K, Long)鍵值對(duì)，其中K為源DStream各個(gè)元素，而Long為該元素出現(xiàn)的次數(shù)。
reduceByKey(func, [numTasks])	如果源DStream 包含的元素為 (K, V) 鍵值對(duì)，則該算子返回一個(gè)新的也包含(K, V)鍵值對(duì)的DStream，其中V是由func聚合得到的。注意：默認(rèn)情況下，該算子使用Spark的默認(rèn)并發(fā)任務(wù)數(shù)（本地模式為2，集群模式下由spark.default.parallelism 決定）。你可以通過(guò)可選參數(shù)numTasks來(lái)指定并發(fā)任務(wù)個(gè)數(shù)。
join(otherStream, [numTasks])	如果源DStream包含元素為(K, V)，同時(shí)otherDStream包含元素為(K, W)鍵值對(duì)，則該算子返回一個(gè)新的DStream，其中源DStream和otherDStream中每個(gè)K都對(duì)應(yīng)一個(gè) (K, (V, W))鍵值對(duì)元素。
cogroup(otherStream, [numTasks])	如果源DStream包含元素為(K, V)，同時(shí)otherDStream包含元素為(K, W)鍵值對(duì)，則該算子返回一個(gè)新的DStream，其中每個(gè)元素類型為包含(K, Seq[V], Seq[W])的tuple。
transform(func)	返回一個(gè)新的DStream，其包含的RDD為源RDD經(jīng)過(guò)func操作后得到的結(jié)果。利用該算子可以對(duì)DStream施加任意的操作。
updateStateByKey(func)	返回一個(gè)包含新”狀態(tài)”的DStream。源DStream中每個(gè)key及其對(duì)應(yīng)的values會(huì)作為func的輸入，而func可以用于對(duì)每個(gè)key的“狀態(tài)”數(shù)據(jù)作任意的更新操作。
?	?

下面我們會(huì)挑幾個(gè)transformation算子深入討論一下。

3.1　updateStateByKey算子

updateStateByKey 算子支持維護(hù)一個(gè)任意的狀態(tài)。要實(shí)現(xiàn)這一點(diǎn)，只需要兩步：

定義狀態(tài) – 狀態(tài)數(shù)據(jù)可以是任意類型。

定義狀態(tài)更新函數(shù) – 定義好一個(gè)函數(shù)，其輸入為數(shù)據(jù)流之前的狀態(tài)和新的數(shù)據(jù)流數(shù)據(jù)，且可其更新步驟1中定義的輸入數(shù)據(jù)流的狀態(tài)。

在每一個(gè)批次數(shù)據(jù)到達(dá)后，Spark都會(huì)調(diào)用狀態(tài)更新函數(shù)，來(lái)更新所有已有key（不管key是否存在于本批次中）的狀態(tài)。如果狀態(tài)更新函數(shù)返回None，則對(duì)應(yīng)的鍵值對(duì)會(huì)被刪除。

舉例如下。假設(shè)你需要維護(hù)一個(gè)流式應(yīng)用，統(tǒng)計(jì)數(shù)據(jù)流中每個(gè)單詞的出現(xiàn)次數(shù)。這里將各個(gè)單詞的出現(xiàn)次數(shù)這個(gè)整型數(shù)定義為狀態(tài)。我們接下來(lái)定義狀態(tài)更新函數(shù)如下：

• Scala
• Java
• Python

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

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

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

該狀態(tài)更新函數(shù)會(huì)為每個(gè)單詞調(diào)用一次，且相應(yīng)的newValues是一個(gè)包含很多個(gè)”1″的數(shù)組（這些1來(lái)自于(word,1)鍵值對(duì)），而runningCount包含之前該單詞的計(jì)數(shù)。本例的完整代碼請(qǐng)參考?StatefulNetworkWordCount.scala。

注意，調(diào)用updateStateByKey前需要配置檢查點(diǎn)目錄，后續(xù)對(duì)此有詳細(xì)的討論，見檢查點(diǎn)（checkpointing）這節(jié)。

3.2　transform算子

transform算子（及其變體transformWith）可以支持任意的RDD到RDD的映射操作。也就是說(shuō)，你可以用tranform算子來(lái)包裝任何DStream API所不支持的RDD算子。例如，將DStream每個(gè)批次中的RDD和另一個(gè)Dataset進(jìn)行關(guān)聯(lián)（join）操作，這個(gè)功能DStream API并沒(méi)有直接支持。不過(guò)你可以用transform來(lái)實(shí)現(xiàn)這個(gè)功能，可見transform其實(shí)為DStream提供了非常強(qiáng)大的功能支持。比如說(shuō)，你可以用事先算好的垃圾信息，對(duì)DStream進(jìn)行實(shí)時(shí)過(guò)濾。

• Scala
• Java
• Python

val spamInfoRDD = ssc.sparkContext.newAPIHadoopRDD(...) // 包含垃圾信息的RDD val cleanedDStream = wordCounts.transform(rdd => {rdd.join(spamInfoRDD).filter(...) // 將DStream中的RDD和spamInfoRDD關(guān)聯(lián)，并實(shí)時(shí)過(guò)濾垃圾數(shù)據(jù) ... })

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

3.3　基于窗口（window）的算子

Spark Streaming同樣也提供基于時(shí)間窗口的計(jì)算，也就是說(shuō)，你可以對(duì)某一個(gè)滑動(dòng)時(shí)間窗內(nèi)的數(shù)據(jù)施加特定tranformation算子。如下圖所示：

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

• （窗口長(zhǎng)度）window length?– 窗口覆蓋的時(shí)間長(zhǎng)度（上圖中為3）
• （滑動(dòng)距離）sliding interval?– 窗口啟動(dòng)的時(shí)間間隔（上圖中為2）

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

下面咱們舉個(gè)栗子。假設(shè)，你需要擴(kuò)展前面的那個(gè)小栗子，你需要每隔10秒統(tǒng)計(jì)一下前30秒內(nèi)的單詞計(jì)數(shù)。為此，我們需要在包含(word, 1)鍵值對(duì)的DStream上，對(duì)最近30秒的數(shù)據(jù)調(diào)用reduceByKey算子。不過(guò)這些都可以簡(jiǎn)單地用一個(gè) reduceByKeyAndWindow搞定。

• Scala
• Java
• Python

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

?

以下列出了常用的窗口算子。所有這些算子都有前面提到的那兩個(gè)參數(shù) – 窗口長(zhǎng)度 和 滑動(dòng)距離。

Transformation窗口算子用途

window(windowLength,?slideInterval)	將源DStream窗口化，并返回轉(zhuǎn)化后的DStream
countByWindow(windowLength,slideInterval)	返回?cái)?shù)據(jù)流在一個(gè)滑動(dòng)窗口內(nèi)的元素個(gè)數(shù)
reduceByWindow(func,?windowLength,slideInterval)	基于數(shù)據(jù)流在一個(gè)滑動(dòng)窗口內(nèi)的元素，用func做聚合，返回一個(gè)單元素?cái)?shù)據(jù)流。func必須滿足結(jié)合律，以便支持并行計(jì)算。
reduceByKeyAndWindow(func,windowLength,?slideInterval, [numTasks])	基于(K, V)鍵值對(duì)DStream，將一個(gè)滑動(dòng)窗口內(nèi)的數(shù)據(jù)進(jìn)行聚合，返回一個(gè)新的包含(K,V)鍵值對(duì)的DStream，其中每個(gè)value都是各個(gè)key經(jīng)過(guò)func聚合后的結(jié)果。 注意：如果不指定numTasks，其值將使用Spark的默認(rèn)并行任務(wù)數(shù)（本地模式下為2，集群模式下由 spark.default.parallelism決定）。當(dāng)然，你也可以通過(guò)numTasks來(lái)指定任務(wù)個(gè)數(shù)。
reduceByKeyAndWindow(func,?invFunc,windowLength,slideInterval, [numTasks])	和前面的reduceByKeyAndWindow() 類似，只是這個(gè)版本會(huì)用之前滑動(dòng)窗口計(jì)算結(jié)果，遞增地計(jì)算每個(gè)窗口的歸約結(jié)果。當(dāng)新的數(shù)據(jù)進(jìn)入窗口時(shí)，這些values會(huì)被輸入func做歸約計(jì)算，而這些數(shù)據(jù)離開窗口時(shí)，對(duì)應(yīng)的這些values又會(huì)被輸入 invFunc 做”反歸約”計(jì)算。舉個(gè)簡(jiǎn)單的例子，就是把新進(jìn)入窗口數(shù)據(jù)中各個(gè)單詞個(gè)數(shù)“增加”到各個(gè)單詞統(tǒng)計(jì)結(jié)果上，同時(shí)把離開窗口數(shù)據(jù)中各個(gè)單詞的統(tǒng)計(jì)個(gè)數(shù)從相應(yīng)的統(tǒng)計(jì)結(jié)果中“減掉”。不過(guò)，你的自己定義好”反歸約”函數(shù)，即：該算子不僅有歸約函數(shù)（見參數(shù)func），還得有一個(gè)對(duì)應(yīng)的”反歸約”函數(shù)（見參數(shù)中的 invFunc）。和前面的reduceByKeyAndWindow() 類似，該算子也有一個(gè)可選參數(shù)numTasks來(lái)指定并行任務(wù)數(shù)。注意，這個(gè)算子需要配置好檢查點(diǎn)（checkpointing）才能用。
countByValueAndWindow(windowLength,slideInterval, [numTasks])	基于包含(K, V)鍵值對(duì)的DStream，返回新的包含(K, Long)鍵值對(duì)的DStream。其中的Long value都是滑動(dòng)窗口內(nèi)key出現(xiàn)次數(shù)的計(jì)數(shù)。 和前面的reduceByKeyAndWindow() 類似，該算子也有一個(gè)可選參數(shù)numTasks來(lái)指定并行任務(wù)數(shù)。
?	?

3.4　Join相關(guān)算子

最后，值得一提的是，你在Spark Streaming中做各種關(guān)聯(lián)（join）操作非常簡(jiǎn)單。

流-流（Stream-stream）關(guān)聯(lián)

一個(gè)數(shù)據(jù)流可以和另一個(gè)數(shù)據(jù)流直接關(guān)聯(lián)。

• • Scala
  • Java
  • Python

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

上面代碼中，stream1的每個(gè)批次中的RDD會(huì)和stream2相應(yīng)批次中的RDD進(jìn)行join。同樣，你可以類似地使用 leftOuterJoin, rightOuterJoin, fullOuterJoin 等。此外，你還可以基于窗口來(lái)join不同的數(shù)據(jù)流，其實(shí)現(xiàn)也很簡(jiǎn)單，如下；）

• Scala
• Java
• Python

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

流-數(shù)據(jù)集（stream-dataset）關(guān)聯(lián)

其實(shí)這種情況已經(jīng)在前面的DStream.transform算子中介紹過(guò)了，這里再舉個(gè)基于滑動(dòng)窗口的例子。

• Scala
• Java
• Python

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

實(shí)際上，在上面代碼里，你可以動(dòng)態(tài)地該表join的數(shù)據(jù)集（dataset）。傳給tranform算子的操作函數(shù)會(huì)在每個(gè)批次重新求值，所以每次該函數(shù)都會(huì)用最新的dataset值，所以不同批次間你可以改變dataset的值。

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

回到頂部

四、DStream輸出算子

輸出算子可以將DStream的數(shù)據(jù)推送到外部系統(tǒng)，如：數(shù)據(jù)庫(kù)或者文件系統(tǒng)。因?yàn)檩敵鏊阕訒?huì)將最終完成轉(zhuǎn)換的數(shù)據(jù)輸出到外部系統(tǒng)，因此只有輸出算子調(diào)用時(shí)，才會(huì)真正觸發(fā)DStream transformation算子的真正執(zhí)行（這一點(diǎn)類似于RDD 的action算子）。目前所支持的輸出算子如下表：

輸出算子用途

print()	在驅(qū)動(dòng)器（driver）節(jié)點(diǎn)上打印DStream每個(gè)批次中的頭十個(gè)元素。 Python API?對(duì)應(yīng)的Python API為?pprint()
saveAsTextFiles(prefix, [suffix])	將DStream的內(nèi)容保存到文本文件。 每個(gè)批次一個(gè)文件，各文件命名規(guī)則為 “prefix-TIME_IN_MS[.suffix]”
saveAsObjectFiles(prefix, [suffix])	將DStream內(nèi)容以序列化Java對(duì)象的形式保存到順序文件中。 每個(gè)批次一個(gè)文件，各文件命名規(guī)則為 “prefix-TIME_IN_MS[.suffix]”Python API?暫不支持Python
saveAsHadoopFiles(prefix, [suffix])	將DStream內(nèi)容保存到Hadoop文件中。 每個(gè)批次一個(gè)文件，各文件命名規(guī)則為 “prefix-TIME_IN_MS[.suffix]”Python API?暫不支持Python
foreachRDD(func)	這是最通用的輸出算子了，該算子接收一個(gè)函數(shù)func，func將作用于DStream的每個(gè)RDD上。 func應(yīng)該實(shí)現(xiàn)將每個(gè)RDD的數(shù)據(jù)推到外部系統(tǒng)中，比如：保存到文件或者寫到數(shù)據(jù)庫(kù)中。 注意，func函數(shù)是在streaming應(yīng)用的驅(qū)動(dòng)器進(jìn)程中執(zhí)行的，所以如果其中包含RDD的action算子，就會(huì)觸發(fā)對(duì)DStream中RDDs的實(shí)際計(jì)算過(guò)程。
?	?

4.1　使用foreachRDD的設(shè)計(jì)模式

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

通常，對(duì)外部系統(tǒng)寫入數(shù)據(jù)需要一些連接對(duì)象（如：遠(yuǎn)程server的TCP連接），以便發(fā)送數(shù)據(jù)給遠(yuǎn)程系統(tǒng)。因此，開發(fā)人員可能會(huì)不經(jīng)意地在Spark驅(qū)動(dòng)器（driver）進(jìn)程中創(chuàng)建一個(gè)連接對(duì)象，然后又試圖在Spark worker節(jié)點(diǎn)上使用這個(gè)連接。如下例所示：

• • Scala
  • Python

dstream.foreachRDD { rdd =>val connection = createNewConnection() // 這行在驅(qū)動(dòng)器（driver）進(jìn)程執(zhí)行rdd.foreach { record =>connection.send(record) // 而這行將在worker節(jié)點(diǎn)上執(zhí)行 } }

這段代碼是錯(cuò)誤的，因?yàn)樗枰堰B接對(duì)象序列化，再?gòu)尿?qū)動(dòng)器節(jié)點(diǎn)發(fā)送到worker節(jié)點(diǎn)。而這些連接對(duì)象通常都是不能跨節(jié)點(diǎn)（機(jī)器）傳遞的。比如，連接對(duì)象通常都不能序列化，或者在另一個(gè)進(jìn)程中反序列化后再次初始化（連接對(duì)象通常都需要初始化，因此從驅(qū)動(dòng)節(jié)點(diǎn)發(fā)到worker節(jié)點(diǎn)后可能需要重新初始化）等。解決此類錯(cuò)誤的辦法就是在worker節(jié)點(diǎn)上創(chuàng)建連接對(duì)象。

然而，有些開發(fā)人員可能會(huì)走到另一個(gè)極端 – 為每條記錄都創(chuàng)建一個(gè)連接對(duì)象，例如：

• Scala
• Python

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

一般來(lái)說(shuō)，連接對(duì)象是有時(shí)間和資源開銷限制的。因此，對(duì)每條記錄都進(jìn)行一次連接對(duì)象的創(chuàng)建和銷毀會(huì)增加很多不必要的開銷，同時(shí)也大大減小了系統(tǒng)的吞吐量。一個(gè)比較好的解決方案是使用 rdd.foreachPartition – 為RDD的每個(gè)分區(qū)創(chuàng)建一個(gè)單獨(dú)的連接對(duì)象，示例如下：

• Scala
• Python

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

?

這樣一來(lái)，連接對(duì)象的創(chuàng)建開銷就攤到很多條記錄上了。

最后，還有一個(gè)更優(yōu)化的辦法，就是在多個(gè)RDD批次之間復(fù)用連接對(duì)象。開發(fā)者可以維護(hù)一個(gè)靜態(tài)連接池來(lái)保存連接對(duì)象，以便在不同批次的多個(gè)RDD之間共享同一組連接對(duì)象，示例如下：

• Scala
• Python

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

?

注意，連接池中的連接應(yīng)該是懶惰創(chuàng)建的，并且有確定的超時(shí)時(shí)間，超時(shí)后自動(dòng)銷毀。這個(gè)實(shí)現(xiàn)應(yīng)該是目前發(fā)送數(shù)據(jù)最高效的實(shí)現(xiàn)方式。

其他要點(diǎn):

• DStream的轉(zhuǎn)化執(zhí)行也是懶惰的，需要輸出算子來(lái)觸發(fā)，這一點(diǎn)和RDD的懶惰執(zhí)行由action算子觸發(fā)很類似。特別地，DStream輸出算子中包含的RDD action算子會(huì)強(qiáng)制觸發(fā)對(duì)所接收數(shù)據(jù)的處理。因此，如果你的Streaming應(yīng)用中沒(méi)有輸出算子，或者你用了dstream.foreachRDD(func)卻沒(méi)有在func中調(diào)用RDD action算子，那么這個(gè)應(yīng)用只會(huì)接收數(shù)據(jù)，而不會(huì)處理數(shù)據(jù)，接收到的數(shù)據(jù)最后只是被簡(jiǎn)單地丟棄掉了。
• 默認(rèn)地，輸出算子只能一次執(zhí)行一個(gè)，且按照它們?cè)趹?yīng)用程序代碼中定義的順序執(zhí)行。

?

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

?

轉(zhuǎn)載于:https://www.cnblogs.com/liuys635/p/11002867.html

總結(jié)

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

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。