介紹
許多應(yīng)用需要處理及時(shí)收到的數(shù)據(jù)，Spark Streaming是Spark為這些應(yīng)用而設(shè)計(jì)的模型祸穷。它允許用戶使用一套和批處理非常接近的 API 來(lái)編寫(xiě)流式計(jì)算應(yīng)用，這樣就可以大量重用批處理應(yīng)用的技術(shù)甚至代碼。
和Spark基于RDD的概念很相似搏屑，Spark Streaming使用離散化流(discretized stream)作為抽象表示溃肪，叫作 DStream免胃。DStream 是隨時(shí)間推移而收到的數(shù)據(jù)的序列。在內(nèi)部乍惊，每個(gè)時(shí)間區(qū)間收到的數(shù)據(jù)都作為 RDD 存在杜秸，而 DStream 是由這些 RDD 所組成的序列(因此得名“離散化”)。DStream 可以從各種輸入源創(chuàng)建润绎，比如 Flume撬碟、Kafka 或者 HDFS。創(chuàng)建出來(lái)的 DStream 支持兩種操作莉撇，一種是轉(zhuǎn)化操作(transformation)呢蛤，會(huì)生成一個(gè)新的DStream，另一種是輸出操作(output operation)棍郎，可以把數(shù)據(jù)寫(xiě)入外部系統(tǒng)中其障。DStream提供了許多與 RDD 所支持的操作相類似的操作支持，還增加了與時(shí)間相關(guān)的新操作涂佃，比如滑動(dòng)窗口励翼。
轉(zhuǎn)化操作
DStream 的轉(zhuǎn)化操作可以分為無(wú)狀態(tài)(stateless)和有狀態(tài)(stateful)兩種。
? 在無(wú)狀態(tài)轉(zhuǎn)化操作中辜荠，每個(gè)批次的處理不依賴于之前批次的數(shù)據(jù)汽抚。第3章和第4章中所講的常見(jiàn)的 RDD 轉(zhuǎn)化操作，例如 map()伯病、filter()造烁、reduceByKey() 等，都是無(wú)狀態(tài)轉(zhuǎn)化操作午笛。
? 相對(duì)地惭蟋，有狀態(tài)轉(zhuǎn)化操作需要使用之前批次的數(shù)據(jù)或者是中間結(jié)果來(lái)計(jì)算當(dāng)前批次的數(shù)據(jù)。有狀態(tài)轉(zhuǎn)化操作包括基于滑動(dòng)窗口的轉(zhuǎn)化操作和追蹤狀態(tài)變化的轉(zhuǎn)化操作药磺。
無(wú)狀態(tài)轉(zhuǎn)換的例子：
- map
- flatmap
- filter
- repartition
- reducebykey
- groupbykey
無(wú)狀態(tài)轉(zhuǎn)化操作也能在多個(gè) DStream 間整合數(shù)據(jù)告组，不過(guò)也是在各個(gè)時(shí)間區(qū)間內(nèi)。例如与涡，鍵值對(duì) DStream 擁有和 RDD 一樣的與連接相關(guān)的轉(zhuǎn)化操作惹谐，也就cogroup()持偏、join()、leftOuterJoin() 等
Transform(): 可以讓你直接操作其內(nèi)部的rdd氨肌。
val outlierDStream = accessLogsDStream.transform { rdd =>
extractOutliers(rdd)
}
有狀態(tài)轉(zhuǎn)換
DStream 的有狀態(tài)轉(zhuǎn)化操作是跨時(shí)間區(qū)間跟蹤數(shù)據(jù)的操作;也就是說(shuō)鸿秆，一些先前批次的數(shù)據(jù)也被用來(lái)在新的批次中計(jì)算結(jié)果。主要的兩種類型是滑動(dòng)窗口和 updateStateByKey()怎囚，前者以一個(gè)時(shí)間階段為滑動(dòng)窗口進(jìn)行操作卿叽，后者則用來(lái)跟蹤每個(gè)鍵的狀態(tài)變化(例如構(gòu)建一個(gè)代表用戶會(huì)話的對(duì)象)。
基于窗口的轉(zhuǎn)化操作
所有基于窗口的操作都需要兩個(gè)參數(shù)恳守，分別為窗口時(shí)長(zhǎng)以及滑動(dòng)步長(zhǎng)考婴，兩者都必須是StreamContext 的批次間隔的整數(shù)倍。窗口時(shí)長(zhǎng)控制每次計(jì)算最近的多少個(gè)批次的數(shù)據(jù)催烘，其實(shí)就是最近的 windowDuration/batchInterval 個(gè)批次沥阱。如果有一個(gè)以 10 秒為批次間隔的源DStream，要?jiǎng)?chuàng)建一個(gè)最近 30 秒的時(shí)間窗口(即最近 3 個(gè)批次)伊群，就應(yīng)當(dāng)把 windowDuration設(shè)為 30 秒考杉。而滑動(dòng)步長(zhǎng)的默認(rèn)值與批次間隔相等，用來(lái)控制對(duì)新的 DStream 進(jìn)行計(jì)算的間隔舰始。如果源 DStream 批次間隔為 10 秒崇棠，并且我們只希望每?jī)蓚€(gè)批次計(jì)算一次窗口結(jié)果，就應(yīng)該把滑動(dòng)步長(zhǎng)設(shè)置為 20 秒丸卷。
val el=errerlines.window(Seconds(3), Seconds(3))
el.count()
盡管可以使用window()寫(xiě)出所有的窗口操作枕稀，Spark Streaming還是提供了一些其他的窗口操作，讓用戶可以高效而方便地使用谜嫉。首先萎坷，reduceByWindow() 和 reduceByKeyAndWindow()讓我們可以對(duì)每個(gè)窗口更高效地進(jìn)行歸約操作。
輸出操作
輸出操作指定了對(duì)流數(shù)據(jù)經(jīng)轉(zhuǎn)化操作得到的數(shù)據(jù)所要執(zhí)行的操作(例如把結(jié)果推入外部數(shù)據(jù)庫(kù)或輸出到屏幕上)沐兰。
print
ipAddressRequestCount.saveAsTextFiles("outputDir", "txt")
saveashadoopfiles
-
Sequence file
val writableIpAddressRequestCount = ipAddressRequestCount.map { (ip, count) => (new Text(ip), new LongWritable(count)) } writableIpAddressRequestCount.saveAsHadoopFiles[ SequenceFileOutputFormat[Text, LongWritable]]("outputDir", "txt") foreachrdd
24/7不間斷運(yùn)行
檢查點(diǎn)機(jī)制
檢查點(diǎn)機(jī)制是我們?cè)赟park Streaming中用來(lái)保障容錯(cuò)性的主要機(jī)制食铐。它可以使SparkStreaming階段性地把應(yīng)用數(shù)據(jù)存儲(chǔ)到諸如HDFS或Amazon S3這樣的可靠存儲(chǔ)系統(tǒng)中，以供恢復(fù)時(shí)使用僧鲁。具體來(lái)說(shuō)，檢查點(diǎn)機(jī)制主要為以下兩個(gè)目的服務(wù)象泵。
- 控制發(fā)生失敗時(shí)需要重算的狀態(tài)數(shù)寞秃。SparkStreaming可以通過(guò)轉(zhuǎn)化圖的譜系圖來(lái)重算狀態(tài)，檢查點(diǎn)機(jī)制則可以控制需要在轉(zhuǎn)化圖中回溯多遠(yuǎn)偶惠。
- 提供驅(qū)動(dòng)器程序容錯(cuò)春寿。如果流計(jì)算應(yīng)用中的驅(qū)動(dòng)器程序崩潰了，你可以重啟驅(qū)動(dòng)器程序并讓驅(qū)動(dòng)器程序從檢查點(diǎn)恢復(fù)忽孽，這樣Spark Streaming就可以讀取之前運(yùn)行的程序處理數(shù)據(jù)的進(jìn)度绑改，并從那里繼續(xù)谢床。
驅(qū)動(dòng)器程序容錯(cuò)
驅(qū)動(dòng)器程序的容錯(cuò)要求我們以特殊的方式創(chuàng)建 StreamingContext。我們需要把檢查點(diǎn)目錄提供給 StreamingContext厘线。與直接調(diào)用 new StreamingContext 不同识腿，應(yīng)該使用StreamingContext.getOrCreate() 函數(shù)。
性能考量
批次和窗口大性熳场（500ms）
并行度
? 增加接收器數(shù)目有時(shí)如果記錄太多導(dǎo)致單臺(tái)機(jī)器來(lái)不及讀入并分發(fā)的話渡讼，接收器會(huì)成為系統(tǒng)瓶頸。這時(shí)你就需要通過(guò)創(chuàng)建多個(gè)輸入 DStream(這樣會(huì)創(chuàng)建多個(gè)接收器)來(lái)增加接收器數(shù)目耳璧，然后使用 union 來(lái)把數(shù)據(jù)合并為一個(gè)數(shù)據(jù)源成箫。
? 將收到的數(shù)據(jù)顯式地重新分區(qū)
如果接收器數(shù)目無(wú)法再增加，你可以通過(guò)使用 DStream.repartition 來(lái)顯式重新分區(qū)輸
入流(或者合并多個(gè)流得到的數(shù)據(jù)流)來(lái)重新分配收到的數(shù)據(jù)旨枯。
? 提高聚合計(jì)算的并行度
對(duì)于像 reduceByKey() 這樣的操作蹬昌，你可以在第二個(gè)參數(shù)中指定并行度。
?
?
?
?
?
?
?
?
?
?
?
?
?
?
?
?
?
?
?
?
?
