Spark入門實(shí)戰(zhàn)系列--7.Spark Streaming（上）--實(shí)時(shí)流計(jì)算Spark Streaming原理介紹 - shishanyuan - 博客園
http://www.cnblogs.com/shishanyuan/p/4747735.html

1、Spark Streaming簡(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的其他子框架，如集群學(xué)習(xí)缰贝、圖計(jì)算等，對(duì)流數(shù)據(jù)進(jìn)行處理畔濒。
Spark Streaming處理的數(shù)據(jù)流圖：

clip_image002

clip_image004

clip_image006

clip_image008

clip_image010

clip_image012

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

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

// Split each line into words
val words = lines.flatMap(_.split(" "))
import org.apache.spark.streaming.StreamingContext._
// Count each word in each batch
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)

// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print()
ssc.start() // Start the computation
ssc.awaitTermination() // Wait for the computation to terminate
1.****創(chuàng)建StreamingContext對(duì)象 同Spark初始化需要?jiǎng)?chuàng)建SparkContext對(duì)象一樣，使用Spark Streaming就需要?jiǎng)?chuàng)建StreamingContext對(duì)象垄琐。創(chuàng)建StreamingContext對(duì)象所需的參數(shù)與SparkContext基本一致边酒，包括指明Master，設(shè)定名稱(如NetworkWordCount)狸窘。需要注意的是參數(shù)Seconds(1)墩朦，Spark Streaming需要指定處理數(shù)據(jù)的時(shí)間間隔，如上例所示的1s翻擒，那么Spark Streaming會(huì)以1s為時(shí)間窗口進(jìn)行數(shù)據(jù)處理氓涣。此參數(shù)需要根據(jù)用戶的需求和集群的處理能力進(jìn)行適當(dāng)?shù)脑O(shè)置牛哺；
2.****創(chuàng)建InputDStream如同Storm的Spout，Spark Streaming需要指明數(shù)據(jù)源劳吠。如上例所示的socketTextStream引润，Spark Streaming以socket連接作為數(shù)據(jù)源讀取數(shù)據(jù)。當(dāng)然Spark Streaming支持多種不同的數(shù)據(jù)源赴背，包括Kafka椰拒、Flume、HDFS/S3凰荚、Kinesis和Twitter等數(shù)據(jù)源；
3.****操作DStream對(duì)于從數(shù)據(jù)源得到的DStream褒脯，用戶可以在其基礎(chǔ)上進(jìn)行各種操作便瑟，如上例所示的操作就是一個(gè)典型的WordCount執(zhí)行流程：對(duì)于當(dāng)前時(shí)間窗口內(nèi)從數(shù)據(jù)源得到的數(shù)據(jù)首先進(jìn)行分割，然后利用Map和ReduceByKey方法進(jìn)行計(jì)算番川，當(dāng)然最后還有使用print()方法輸出結(jié)果到涂；
4.****啟動(dòng)Spark Streaming之前所作的所有步驟只是創(chuàng)建了執(zhí)行流程，程序沒有真正連接上數(shù)據(jù)源颁督，也沒有對(duì)數(shù)據(jù)進(jìn)行任何操作践啄，只是設(shè)定好了所有的執(zhí)行計(jì)劃，當(dāng)ssc.start()啟動(dòng)后程序才真正進(jìn)行所有預(yù)期的操作沉御。
至此對(duì)于Spark Streaming的如何使用有了一個(gè)大概的印象屿讽，在后面的章節(jié)我們會(huì)通過源代碼深入探究一下Spark Streaming的執(zhí)行流程。
2.2.2 DStream的輸入源
在Spark Streaming中所有的操作都是基于流的吠裆，而輸入源是這一系列操作的起點(diǎn)伐谈。輸入 DStreams 和 DStreams 接收的流都代表輸入數(shù)據(jù)流的來(lái)源，在Spark Streaming 提供兩種內(nèi)置數(shù)據(jù)流來(lái)源：
l 基礎(chǔ)來(lái)源 在 StreamingContext API 中直接可用的來(lái)源试疙。例如：文件系統(tǒng)诵棵、Socket（套接字）連接和 Akka actors；
l 高級(jí)來(lái)源 如 Kafka祝旷、Flume履澳、Kinesis、Twitter 等怀跛，可以通過額外的實(shí)用工具類創(chuàng)建距贷。
2.2.2.1 基礎(chǔ)來(lái)源
在前面分析怎樣使用Spark Streaming的例子中我們已看到ssc.socketTextStream()方法，可以通過 TCP 套接字連接敌完，從從文本數(shù)據(jù)中創(chuàng)建了一個(gè) DStream储耐。除了套接字，StreamingContext 的API還提供了方法從文件和 Akka actors 中創(chuàng)建DStreams作為輸入源滨溉。
Spark Streaming提供了streamingContext.fileStream(dataDirectory)方法可以從任何文件系統(tǒng)(如：HDFS什湘、S3长赞、NFS等）的文件中讀取數(shù)據(jù)，然后創(chuàng)建一個(gè)DStream闽撤。Spark Streaming 監(jiān)控 dataDirectory 目錄和在該目錄下任何文件被創(chuàng)建處理(不支持在嵌套目錄下寫文件)得哆。需要注意的是：讀取的必須是具有相同的數(shù)據(jù)格式的文件；創(chuàng)建的文件必須在dataDirectory 目錄下哟旗，并通過自動(dòng)移動(dòng)或重命名成數(shù)據(jù)目錄贩据；文件一旦移動(dòng)就不能被改變，如果文件被不斷追加,新的數(shù)據(jù)將不會(huì)被閱讀闸餐。對(duì)于簡(jiǎn)單的文本文饱亮，可以使用一個(gè)簡(jiǎn)單的方法streamingContext.textFileStream(dataDirectory)來(lái)讀取數(shù)據(jù)。
Spark Streaming也可以基于自定義 Actors 的流創(chuàng)建DStream 舍沙，通過 Akka actors 接受數(shù)據(jù)流近上，使用方法streamingContext.actorStream(actorProps, actor-name)。Spark Streaming使用streamingContext.queueStream(queueOfRDDs)方法可以創(chuàng)建基于 RDD 隊(duì)列的DStream拂铡，每個(gè)RDD 隊(duì)列將被視為DStream 中一塊數(shù)據(jù)流進(jìn)行加工處理壹无。
2.2.2.2 高級(jí)來(lái)源
這一類的來(lái)源需要外部 non-Spark 庫(kù)的接口，其中一些有復(fù)雜的依賴關(guān)系(如 Kafka感帅、Flume)斗锭。因此通過這些來(lái)源創(chuàng)建DStreams 需要明確其依賴。例如失球，如果想創(chuàng)建一個(gè)使用 Twitter tweets 的數(shù)據(jù)的DStream 流岖是，必須按以下步驟來(lái)做：
1）在 SBT 或 Maven工程里添加 spark-streaming-twitter_2.10 依賴。
2）開發(fā)：導(dǎo)入 TwitterUtils 包她倘，通過 TwitterUtils.createStream 方法創(chuàng)建一個(gè)DStream璧微。
3）部署：添加所有依賴的 jar 包(包括依賴的spark-streaming-twitter_2.10 及其依賴)，然后部署應(yīng)用程序硬梁。
需要注意的是前硫，這些高級(jí)的來(lái)源一般在Spark Shell中不可用，因此基于這些高級(jí)來(lái)源的應(yīng)用不能在Spark Shell中進(jìn)行測(cè)試荧止。如果你必須在Spark shell中使用它們屹电，你需要下載相應(yīng)的Maven工程的Jar依賴并添加到類路徑中。
其中一些高級(jí)來(lái)源如下：
lTwitter Spark Streaming的TwitterUtils工具類使用Twitter4j跃巡，Twitter4J 庫(kù)支持通過任何方法提供身份驗(yàn)證信息赁咙，你可以得到公眾的流歌殃，或得到基于關(guān)鍵詞過濾流援所。
lFlume Spark Streaming可以從Flume中接受數(shù)據(jù)不狮。
lKafka Spark Streaming可以從Kafka中接受數(shù)據(jù)。
lKinesis Spark Streaming可以從Kinesis中接受數(shù)據(jù)。
需要重申的一點(diǎn)是在開始編寫自己的 SparkStreaming 程序之前偷线，一定要將高級(jí)來(lái)源依賴的Jar添加到SBT 或 Maven 項(xiàng)目相應(yīng)的artifact中磨确。常見的輸入源和其對(duì)應(yīng)的Jar包如下圖所示。

clip_image014

描述

map(func)

源 DStream的每個(gè)元素通過函數(shù)func返回一個(gè)新的DStream骗炉。

flatMap(func)

類似與map操作，不同的是每個(gè)輸入元素可以被映射出0或者更多的輸出元素材失。

filter(func)

在源DSTREAM上選擇Func函數(shù)返回僅為true的元素,最終返回一個(gè)新的DSTREAM 痕鳍。

repartition(numPartitions)

通過輸入的參數(shù)numPartitions的值來(lái)改變DStream的分區(qū)大小。

union(otherStream)

返回一個(gè)包含源DStream與其他 DStream的元素合并后的新DSTREAM龙巨。

count()

對(duì)源DStream內(nèi)部的所含有的RDD的元素?cái)?shù)量進(jìn)行計(jì)數(shù)，返回一個(gè)內(nèi)部的RDD只包含一個(gè)元素的DStreaam熊响。

reduce(func)

使用函數(shù)func（有兩個(gè)參數(shù)并返回一個(gè)結(jié)果）將源DStream 中每個(gè)RDD的元素進(jìn)行聚 合操作,返回一個(gè)內(nèi)部所包含的RDD只有一個(gè)元素的新DStream旨别。

countByValue()

計(jì)算DStream中每個(gè)RDD內(nèi)的元素出現(xiàn)的頻次并返回新的DStream[(K,Long)]，其中K是RDD中元素的類型汗茄，Long是元素出現(xiàn)的頻次秸弛。

reduceByKey(func, [numTasks])

當(dāng)一個(gè)類型為（K，V）鍵值對(duì)的DStream被調(diào)用的時(shí)候,返回類型為類型為（K洪碳，V）鍵值對(duì)的新 DStream,其中每個(gè)鍵的值V都是使用聚合函數(shù)func匯總递览。注意：默認(rèn)情況下，使用 Spark的默認(rèn)并行度提交任務(wù)（本地模式下并行度為2瞳腌，集群模式下位8）绞铃，可以通過配置numTasks設(shè)置不同的并行任務(wù)數(shù)。

join(otherStream, [numTasks])

當(dāng)被調(diào)用類型分別為（K嫂侍，V）和（K儿捧，W）鍵值對(duì)的2個(gè)DStream時(shí)，返回類型為（K挑宠，（V菲盾，W））鍵值對(duì)的一個(gè)新 DSTREAM。

cogroup(otherStream, [numTasks])

當(dāng)被調(diào)用的兩個(gè)DStream分別含有(K, V) 和(K, W)鍵值對(duì)時(shí),返回一個(gè)(K, Seq[V], Seq[W])類型的新的DStream各淀。

transform(func)

通過對(duì)源DStream的每RDD應(yīng)用RDD-to-RDD函數(shù)返回一個(gè)新的DStream懒鉴，這可以用來(lái)在DStream做任意RDD操作。

updateStateByKey(func)

返回一個(gè)新狀態(tài)的DStream,其中每個(gè)鍵的狀態(tài)是根據(jù)鍵的前一個(gè)狀態(tài)和鍵的新值應(yīng)用給定函數(shù)func后的更新碎浇。這個(gè)方法可以被用來(lái)維持每個(gè)鍵的任何狀態(tài)數(shù)據(jù)临谱。

在上面列出的這些操作中璃俗，transform()方法和updateStateByKey()方法值得我們深入的探討一下：
l transform(func)****操作
該transform操作（轉(zhuǎn)換操作）連同其其類似的 transformWith操作允許DStream 上應(yīng)用任意RDD-to-RDD函數(shù)。它可以被應(yīng)用于未在 DStream API 中暴露任何的RDD操作吴裤。例如旧找，在每批次的數(shù)據(jù)流與另一數(shù)據(jù)集的連接功能不直接暴露在DStream API 中，但可以輕松地使用transform操作來(lái)做到這一點(diǎn)麦牺，這使得DStream的功能非常強(qiáng)大钮蛛。例如，你可以通過連接預(yù)先計(jì)算的垃圾郵件信息的輸入數(shù)據(jù)流（可能也有Spark生成的）剖膳，然后基于此做實(shí)時(shí)數(shù)據(jù)清理的篩選魏颓，如下面官方提供的偽代碼所示。事實(shí)上吱晒，也可以在transform方法中使用機(jī)器學(xué)習(xí)和圖形計(jì)算的算法甸饱。
l updateStateByKey****操作
該 updateStateByKey 操作可以讓你保持任意狀態(tài)，同時(shí)不斷有新的信息進(jìn)行更新仑濒。要使用此功能叹话，必須進(jìn)行兩個(gè)步驟 ：
（1） 定義狀態(tài) - 狀態(tài)可以是任意的數(shù)據(jù)類型。
（2） 定義狀態(tài)更新函數(shù) - 用一個(gè)函數(shù)指定如何使用先前的狀態(tài)和從輸入流中獲取的新值 更新狀態(tài)墩瞳。
讓我們用一個(gè)例子來(lái)說(shuō)明驼壶，假設(shè)你要進(jìn)行文本數(shù)據(jù)流中單詞計(jì)數(shù)。在這里喉酌，正在運(yùn)行的計(jì)數(shù)是狀態(tài)而且它是一個(gè)整數(shù)热凹。我們定義了更新功能如下：

clip_image016

clip_image018

描述

window(windowLength, slideInterval)

返回一個(gè)基于源DStream的窗口批次計(jì)算后得到新的DStream止状。

countByWindow(windowLength,slideInterval)

返回基于滑動(dòng)窗口的DStream中的元素的數(shù)量。

reduceByWindow(func, windowLength,slideInterval)

基于滑動(dòng)窗口對(duì)源DStream中的元素進(jìn)行聚合操作攒霹，得到一個(gè)新的DStream怯疤。

reduceByKeyAndWindow(func,windowLength,slideInterval, [numTasks])

基于滑動(dòng)窗口對(duì)（K，V）鍵值對(duì)類型的DStream中的值按K使用聚合函數(shù)func進(jìn)行聚合操作催束，得到一個(gè)新的DStream集峦。

reduceByKeyAndWindow(func, invFunc,windowLength,slideInterval, [numTasks])

一個(gè)更高效的reduceByKkeyAndWindow()的實(shí)現(xiàn)版本，先對(duì)滑動(dòng)窗口中新的時(shí)間間隔內(nèi)數(shù)據(jù)增量聚合并移去最早的與新增數(shù)據(jù)量的時(shí)間間隔內(nèi)的數(shù)據(jù)統(tǒng)計(jì)量。例如塔淤，計(jì)算t+4秒這個(gè)時(shí)刻過去5秒窗口的WordCount摘昌，那么我們可以將t+3時(shí)刻過去5秒的統(tǒng)計(jì)量加上[t+3，t+4]的統(tǒng)計(jì)量高蜂，在減去[t-2聪黎，t-1]的統(tǒng)計(jì)量，這種方法可以復(fù)用中間三秒的統(tǒng)計(jì)量备恤，提高統(tǒng)計(jì)的效率稿饰。

countByValueAndWindow(windowLength,slideInterval, [numTasks])

基于滑動(dòng)窗口計(jì)算源DStream中每個(gè)RDD內(nèi)每個(gè)元素出現(xiàn)的頻次并返回DStream[(K,Long)]，其中K是RDD中元素的類型露泊，Long是元素頻次喉镰。與countByValue一樣，reduce任務(wù)的數(shù)量可以通過一個(gè)可選參數(shù)進(jìn)行配置惭笑。

clip_image020

描述

print()

在Driver中打印出DStream中數(shù)據(jù)的前10個(gè)元素。

saveAsTextFiles(prefix, [suffix])

將DStream中的內(nèi)容以文本的形式保存為文本文件盆赤，其中每次批處理間隔內(nèi)產(chǎn)生的文件以prefix-TIME_IN_MS[.suffix]的方式命名贾富。

saveAsObjectFiles(prefix, [suffix])

將DStream中的內(nèi)容按對(duì)象序列化并且以SequenceFile的格式保存。其中每次批處理間隔內(nèi)產(chǎn)生的文件以prefix-TIME_IN_MS[.suffix]的方式命名牺六。

saveAsHadoopFiles(prefix, [suffix])

將DStream中的內(nèi)容以文本的形式保存為Hadoop文件颤枪，其中每次批處理間隔內(nèi)產(chǎn)生的文件以prefix-TIME_IN_MS[.suffix]的方式命名。

foreachRDD(func)

最基本的輸出操作淑际，將func函數(shù)應(yīng)用于DStream中的RDD上畏纲，這個(gè)操作會(huì)輸出數(shù)據(jù)到外部系統(tǒng)，比如保存RDD到文件或者網(wǎng)絡(luò)數(shù)據(jù)庫(kù)等春缕。需要注意的是func函數(shù)是在運(yùn)行該streaming應(yīng)用的Driver進(jìn)程里執(zhí)行的盗胀。

dstream.foreachRDD是一個(gè)非常強(qiáng)大的輸出操作，它允將許數(shù)據(jù)輸出到外部系統(tǒng)锄贼。但是 票灰，如何正確高效地使用這個(gè)操作是很重要的，下面展示了如何去避免一些常見的錯(cuò)誤宅荤。
通常將數(shù)據(jù)寫入到外部系統(tǒng)需要?jiǎng)?chuàng)建一個(gè)連接對(duì)象（如 TCP連接到遠(yuǎn)程服務(wù)器）屑迂，并用它來(lái)發(fā)送數(shù)據(jù)到遠(yuǎn)程系統(tǒng)。出于這個(gè)目的冯键，開發(fā)者可能在不經(jīng)意間在Spark driver端創(chuàng)建了連接對(duì)象惹盼，并嘗試使用它保存RDD中的記錄到Spark worker上，如下面代碼：

clip_image022

clip_image024

參考資料：
（1）《Spark Streaming》 http://blog.debugo.com/spark-streaming/