基本概念

接下來，我們在上一節例子的基礎上，來闡述Spark Streaming的基本知識。

連結

和Spark類似，Spark Streaming也包含在maven的中央倉庫中。為了寫基於Spark Streaming的程式，你需要為你的SBT或Maven工程分別新增以下依懶:

Maven:

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

SBT:

libraryDependencies += "org.apache.spark" % "spark-streaming_2.10" % "1.4.0"

為了從Kafka, Flume, 和 Kinesis等源中提取資料，你需要新增額外的依懶。因為Spark Streaming的核心API並不直接支援。下面是一些常用的依懶：

要獲得最新的列表，請在maven中央倉庫Maven repository檢視支援的源和座標。

初始化StreamingContext

為了初始化一個Spark Streaming程式，需要建立一個StreamingContext物件，它是所有Spark Streaming功能的主入口點。

可以從一個SparkConf物件建立一個StreamingContext

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引數是你的應用程式在叢集UI上的名字。master是Spark, Mesos或者YARN cluster 的URL,或者用"local[*]'字串表示執行在本地模式。

在實踐中，當在叢集上執行程式時，你不希望硬編碼master的值。你可以通過從spark-submit啟動你的程式，並指明引數。然而，為了本地測試或者單元測試，你可以傳遞"local[*]”引數來執行Spark Streaming程式。注意，你可以通過ssc.SparkContext的方式來訪問SparkContext.

批處理的間隔需要根據你的應用程式的延遲要求和可用的叢集資源來設定。

SparkStreaming物件也可以由已經存在的SparkContext來建立:

import org.apache.spark.streaming._

val sc = ...                // existing SparkContext
val ssc = new StreamingContext(sc, Seconds(1))

在建立了context之後，你需要做以下幾步：

1.通過建立輸入DStreams來定義輸入源

2.通過對DStreams應用轉變操作和輸出操作來定義流計算。

3.用streamingContext.start()來開始接收資料和處理流程。

4.通過streamingContext.awaitTermination()方法來等待處理結束（手動結束或因為錯誤）

5.還可以通過streamingContext.stop()來手動結束程序。

需要記住的關鍵點：

1.一旦context啟動後，就不能再新增新的streaming計算或操作。

2.一旦context停止後，就不能再重新啟動它了。

3.在同一時間內，一個JVM上只能執行一個StreamingContext

4.在StreamingContext上的stop()操作也會將SparkContext停止。如果只想停止StreamingContext，對stop的可選引數stopSparkContext設定為false.

5.一個SparkContext可以用來建立多個StreamingContext,只要前一個StreamingContext已經停止了。