spark Dstreams-基本內(nèi)容

IT那活兒發(fā)布于2023-01-11 13:20 / 1491人閱讀

點(diǎn)擊上方“IT那活兒”，關(guān)注后了解更多內(nèi)容，不管IT什么活兒，干就完了！！！

簡介

Spark Streaming是spark core的延伸，支持實(shí)時(shí)數(shù)據(jù)流的可伸縮、高吞吐量、容錯(cuò)機(jī)制，支持的數(shù)據(jù)源有kafka、Kinesis、TCP等），且可以使用復(fù)雜的算法（map,reduce,join,window等）處理數(shù)據(jù)，最后可以將處理后的數(shù)據(jù)推送到文件系統(tǒng)、數(shù)據(jù)庫或儀表盤。（如下圖，您也可以再數(shù)據(jù)流上應(yīng)用spark的機(jī)器學(xué)習(xí)和圖形算法。）

01

在內(nèi)部，spark streaming接受實(shí)時(shí)數(shù)據(jù)流并且將數(shù)據(jù)流切割成一個(gè)個(gè)的小批次，然后由spark引擎進(jìn)行處理，以批量生成最終結(jié)果流。

Spark streaming提供一個(gè)高級(jí)抽象叫做離散化數(shù)據(jù)流Dstreams，他代表一個(gè)連續(xù)的數(shù)據(jù)流，可以從kafka，kinesis等源輸入創(chuàng)建數(shù)據(jù)流，也可以通過對(duì)其他數(shù)據(jù)流應(yīng)用高級(jí)操作來創(chuàng)建數(shù)據(jù)流。

在內(nèi)部，一個(gè)Dstreams代表一系列的RDD。

02

具體步驟：

批處理間隔必須根據(jù)應(yīng)用程序和可用集群資源進(jìn)行設(shè)置，ssc也可以通過SparkContext創(chuàng)建：

val ssc = new StreamingContext(sc, Seconds(1))

創(chuàng)建ssc之后，按照下面步驟進(jìn)行：

通過輸入數(shù)據(jù)源創(chuàng)建Dstreams。
對(duì)數(shù)據(jù)流進(jìn)行轉(zhuǎn)換和輸出操作。
使用ssc.start()啟動(dòng)應(yīng)用程序。
使用ssc.awaitTermination()等待停止處理。（手動(dòng)或由于任何錯(cuò)誤）
使用ssc.stop()手動(dòng)停止應(yīng)用程序。

舉例演示：

使用spark-streaming前提是添加依賴：

如果數(shù)據(jù)源為kafka，由于spark core中無kafka依賴，需要額外添加依賴：

StreamingContext是流功能的入口，我們創(chuàng)建一個(gè)本地StreamingContext，兩個(gè)執(zhí)行線程，批處理間隔為1秒。

val conf = new SparkConf().setMaster("local[2]").setAppName("Dstreams")

master是spark,Mesos,Kubernetes，或者YARN集群URL，或者本地模式運(yùn)行的特殊local[*]字符串。

實(shí)際上，在集群上運(yùn)行時(shí)，你不希望在程序中寫死m(xù)aster，而是在使用spark-submit啟動(dòng)應(yīng)用程序時(shí)在那里接收master。

03

離散化數(shù)據(jù)流（Dstreams）

Dstreams是spark streaming提供的基本抽象，它代表一個(gè)連續(xù)的數(shù)據(jù)流，不管是輸入源創(chuàng)建的Dsteam還是通過轉(zhuǎn)換生成的輸入流，在內(nèi)部Dstream代表一系列連續(xù)的RDD，每個(gè)RDD都是特定時(shí)間間隔的數(shù)據(jù)。

應(yīng)用于數(shù)據(jù)流的任何操作都轉(zhuǎn)化為底層RDD的操作，例如在之前的例子中，對(duì)用數(shù)據(jù)流中每個(gè)RDD應(yīng)用flatMap操作，如圖：

這些底層RDD轉(zhuǎn)換由spark引擎計(jì)算，Dstream操作隱藏了大部分這些細(xì)節(jié)，并為開發(fā)人員提供了更高級(jí)別的API。

04

輸入數(shù)據(jù)流（Input Dstreams）表示從源接受的輸入數(shù)據(jù)流。

在上面的例子中，lines就是輸入數(shù)據(jù)流（Input DStreams），代表從TCP socket接收的數(shù)據(jù)。

每個(gè)輸入數(shù)據(jù)流（Input DStreams）除了文件流都與接收器對(duì)象關(guān)聯(lián)，該對(duì)象從源接受的數(shù)據(jù)并將其在spark的內(nèi)存中進(jìn)行處理。

Spark streaming提供兩類內(nèi)置streaming源：

基本source：在StreamingContext中可以直接用的基本source，像file source，socket source。
高級(jí)source：像kafka，Kinesis等，需要添加額外依賴。

注意：

一旦StreamingContext啟動(dòng)，就不能設(shè)置或添加新的流計(jì)算。
StreamingContext一旦停止就不能重啟。
一個(gè)JVM中只能同時(shí)只能有一個(gè)StreamingContext，ssc.stop()也會(huì)停止SparkContext，如果僅停止streamingContext，需要ssc.stop(false)。
只要在創(chuàng)建下一個(gè)StreamingContext之前停止上一個(gè)StreamingContext而不停止SparkContext，就可以重新用于創(chuàng)建StreamingContetx。
如果希望在應(yīng)用程序中并行接受多個(gè)數(shù)據(jù)流，則可以創(chuàng)建多個(gè)輸入數(shù)據(jù)流，這將同時(shí)創(chuàng)建接受多個(gè)輸入數(shù)據(jù)流的多個(gè)接收器，Spark worker/executor是一個(gè)長期運(yùn)行的任務(wù)，因此它占用分配給Spark流應(yīng)用程序的一個(gè)核心。
因此，務(wù)必記住，Spark Streaming應(yīng)用程序需要分配足夠的內(nèi)核（或線程，如果在本地運(yùn)行）來處理接收到的數(shù)據(jù)，以及運(yùn)行接收器。
在本地運(yùn)行spark streaming應(yīng)用程序時(shí)，不要將master設(shè)置成local[1]或local，這兩種情況意味著只有一個(gè)線程用于本地運(yùn)行任務(wù)，如果您使用的是基于接收器的輸入數(shù)據(jù)流（例如套接字，kafka等），則使用單個(gè)線程運(yùn)行接收器，就沒有其他線程處理數(shù)據(jù)了。
在集群上運(yùn)行時(shí)，分配給spark streaming應(yīng)用程序的內(nèi)核數(shù)必須大于接收器數(shù)，否則應(yīng)用只能接受數(shù)據(jù)而無法處理數(shù)據(jù)。