Spark Streaming
大綱：java
-
Spark Streaming簡介node
-
Spark Streaming的原理和架構python
-
Spark Streaming之基礎抽象DStream算法
-
DStream相關操做數據庫
-
Spark Streaming與flume整合apache
-
Spark Streaming與kafka整合編程
第一章 Spark Streaming概述
1.1 Spark Streaming簡介
Spark Streaming能夠很容易的構建高吞吐量和容錯能力強的流式應用.Spark Streaming相似於Apache Storm，用於流式數據的處理。根據其官方文檔介紹，Spark Streaming有高吞吐量和容錯能力強等特色。Spark Streaming支持的數據源有不少，例如：Kafka、Flume、Twitter、ZeroMQ和簡單的TCP套接字等等。數據輸入後能夠用Spark的高度抽象操做如：map、reduce、join、window等進行運算。而結果也能保存在不少地方，如HDFS，數據庫等。另外Spark Streaming也能和MLlib（機器學習）以及Graphx完美融合。api
Spark Streaming是核心Spark Core API的擴展,能夠實現實時數據流的可擴展,高吞吐,容錯流處理.數據能夠來自不少數據源(例如kafka,Flume或者TCP套接字)中獲取,而且可使用高級函數表示的複雜算法進行處理.最後,處理後的數據能夠推送到文件系統,數據庫和實時儀表板.緩存
1.2 Spark Streaming的特色
1.2.1 易用性
能夠像編寫離線批處理同樣去編寫流式程序，支持java/scala/python語言。服務器
1.2.2 容錯性
Spark Streaming在沒有額外代碼和配置的狀況下能夠恢復丟失的工做。
1.2.3 易整合
流式處理與批處理和交互式查詢相結合。
1.3 Spark Streaming與Storm的對比
| Spark Streaming | Storm |
|---|---|
|
|
| 開發語言:Scala/Java/Python | 開發語言:Clojure |
| 編程模型:DStream | 編程模型:Spout/Bolt |
第二章 Spark Streaming原理
2.1 Spark Streaming原理
Spark Streaming是基於spark的流式批處理引擎,其基本原理是把輸入數據以某一段時間間隔批量的處理,當批處理間隔縮短到秒級時,即可以用於處理實時數據流.也就是說,在內部,它的工做原理以下圖所示,Spark Streaming接收實時輸入數據流並將數據分紅批處理,而後由Spark引擎處理,以批量生成最終結果流
2.2 Spark Streaming計算流程
Spark Streaming是將流式計算分解成一系列短小的批處理做業.這裏的批處理引擎是Spark Core.也就是把Spark Streaming的輸入數據按照batch size(如1秒)分紅一段一段的數據(Discretized Stream),每一段數據都轉換成Spark中的RDD(Resilient Distributed Dataset),而後將Spark Streaming中對DStream的transformation操做變爲針對Spark中對RDD的transformation操做,將RDD通過操做變成中間結果保存在內存中.整個流式計算根據業務的需求能夠對中間的結果進行緩存或者存儲到外部設備.下圖顯示了Spark Streaming的整個流程。
2.3 Spark Streaming的容錯性
對於流式計算來講，容錯性相當重要。首先咱們要明確一下Spark中RDD的容錯機制。每個RDD都是一個不可變的分佈式可重算的數據集，其記錄着肯定性的操做繼承關係（lineage），因此只要輸入數據是可容錯的，那麼任意一個RDD的分區（Partition）出錯或不可用，都是能夠利用原始輸入數據經過轉換操做而從新算出的。對於Spark Streaming來講，其RDD的傳承關係以下圖所示：
圖中的每個橢圓形表示一個RDD，橢圓形中的每一個圓形表明一個RDD中的一個Partition，圖中的每一列的多個RDD表示一個DStream（圖中有三個DStream），而每一行最後一個RDD則表示每個Batch Size所產生的中間結果RDD。咱們能夠看到圖中的每個RDD都是經過lineage相鏈接的，因爲Spark Streaming輸入數據能夠來自於磁盤，例如HDFS（多份拷貝）或是來自於網絡的數據流（Spark Streaming會將網絡輸入數據的每個數據流拷貝兩份到其餘的機器）都能保證容錯性，因此RDD中任意的Partition出錯，均可以並行地在其餘機器上將缺失的Partition計算出來。這個容錯恢復方式比連續計算模型（如Storm）的效率更高。
2.4 Spark Streaming實時性
對於實時性的討論，會牽涉到流式處理框架的應用場景。Spark Streaming將流式計算分解成多個Spark Job，對於每一段數據的處理都會通過Spark DAG圖分解以及Spark的任務集的調度過程。對於目前版本的Spark Streaming而言，其最小的Batch Size的選取在0.5~2秒鐘之間（Storm目前最小的延遲是100ms左右），因此Spark Streaming可以知足除對實時性要求很是高（如高頻實時交易）以外的全部流式準實時計算場景。
第三章 DStream
3.1 什麼是DStream
Discretized Stream是Spark Streaming的基礎抽象，表明持續性的數據流和通過各類Spark算子操做後的結果數據流。在內部實現上，DStream是一系列連續的RDD來表示。每一個RDD含有一段時間間隔內的數據，以下圖：
應用於DStream的任何操做都轉換爲底層RDD上的操做,例如,在以前將lines DStream轉換爲words DStream,flatMap操做應用於lines DStream中的每個RDD以生成words DStream的words RDD
這些底層RDD轉換是由Spark引擎計算的,DStream操做隱藏了大部分細節.
3.2 DStream相關操做
DStream上的操做與RDD的相似,分爲Transformations(轉換)和OutputOperations(輸出)兩種,此外轉換操做中還有一些比較特殊的操做,如"updateStateByKey(),transform()以及各類Window相關的操做.
3.2.1 Transformations on DStreams
| Transformation | Meaning |
|---|---|
| map(func) | 對DStream中的各個元素進行func函數操做，而後返回一個新的DStream |
| flatMap(func) | 與map方法相似，只不過各個輸入項能夠被輸出爲零個或多個輸出項 |
| filter(func) | 過濾出全部函數func返回值爲true的DStream元素並返回一個新的DStream |
| repartition(numPartitions) | 增長或減小DStream中的分區數，從而改變DStream的並行度 |
| union(otherStream) | 將源DStream和輸入參數爲otherDStream的元素合併，並返回一個新的DStream. |
| count() | 經過對DStream中的各個RDD中的元素進行計數，而後返回只有一個元素的RDD構成的DStream |
| reduce(func) | 對源DStream中的各個RDD中的元素利用func進行聚合操做，而後返回只有一個元素的RDD構成的新的DStream. |
| countByValue() | 對於元素類型爲K的DStream，返回一個元素爲（K,Long）鍵值對形式的新的DStream，Long對應的值爲源DStream中各個RDD的key出現的次數 |
| reduceByKey(func[numTasks]) | 利用func函數對源DStream中的key進行聚合操做，而後返回新的（K，V）對構成的DStream |
| join(otherStream,[numTasks]) | 輸入爲（K,V)、（K,W）類型的DStream，返回一個新的（K，（V，W））類型的DStream |
| cogroup(otherStream,[numTasks]) | 輸入爲（K,V)、（K,W）類型的DStream，返回一個新的(K,Seq[V], Seq[W]) 元組類型的DStream |
| transform(func) | 經過RDD-to-RDD函數做用於DStream中的各個RDD，能夠是任意的RDD操做，從而返回一個新的RDD |
| updateStateByKey(func) | 根據key的以前狀態值和key的新值，對key進行更新，返回一個新狀態的DStream |
3.2.2 特殊的Transformations
3.2.2.1 UpdateStateByKey Operation
UpdateStateByKey用於記錄歷史記錄，保存上次的狀態.若是使用UpdateStateByKey,必須執行兩個步驟
-
定義狀態,狀態能夠是任意數據類型.
-
定義狀態更新函數.使用函數指定如何使用先前狀態和輸入流中的新值更新狀態.
3.2.2.2 Window Operation(開窗函數)
滑動窗口轉換操做的計算過程以下圖所示，咱們能夠事先設定一個滑動窗口的長度（也就是窗口的持續時間），而且設定滑動窗口的時間間隔（每隔多長時間執行一次計算），而後，就可讓窗口按照指定時間間隔在源DStream上滑動，每次窗口停放的位置上，都會有一部分DStream被框入窗口內，造成一個小段的DStream，這時，就能夠啓動對這個小段DStream的計算。
-
紅色的矩形就是一個窗口，窗口框住的是一段時間內的數據流。
-
這裏面每個time都是時間單元，在官方的例子中，每隔window size是3 time unit, 並且每隔2個單位時間，窗口會slide一次。
-
因此基於窗口的操做,須要指定2個參數:
-
window length - The duration of the window (3 in the figure).
-
窗口長度,一段時間內數據的容器.
-
-
sliding interval - The interval at which the window operation is performed (2 in the figure).
-
滑動間隔,每隔多久計算一次.
-
-
3.2.3 Ouput Operations on DStreams
Output Operations能夠將DStream的數據輸出到外部的數據庫或文件系統，當某個Output Operations被調用時（與RDD的Action相同），Spark Streaming程序纔會開始真正的計算過程。
| Output Operation | Meaning |
|---|---|
| print() | 打印到控制檯 |
| saveAsTextFiles(prefix,[suffix]) | 保存流的內容爲文本文件，文件名爲"prefix-TIME_IN_MS[.suffix]". |
| saveAsObjectFiles(prefix,[suffix]) | 保存流的內容爲SequenceFile，文件名爲"prefix-TIME_IN_MS[.suffix]". |
| saveAsHadoopFiles(prefix,[suffix]) | 保存流的內容爲hadoop文件，文件名爲"prefix-TIME_IN_MS[.suffix]". |
| foreachRDD(func) | 對Dstream裏面的每一個RDD執行func |
第四章 DStream操做實戰
4.1 SparkStreaming接受socket數據,實現詞頻統計
4.1.1 架構圖
4.1.2 代碼實現
4.1.2.1 安裝netcat
首先在Linux服務器用YUM安裝nc工具,nc命令式netcat命令的簡稱,它式用來設置路由器,咱們能夠利用它向某個端口發送數據.執行以下命令:
nc -lk 9999
4.1.2.2 編寫Spark Streaming程序
導入Spark Streaming的pom配置
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.11</artifactId>
<version>2.2.0</version>
</dependency>
Java語言
/**
* @author dw
* @date 2018-08-24 16:39:39
* @description dw
* 使用Spark Streaming讀取Socket數據並實現詞頻統計
*/
public class SparkStreamingFromSocketWithJava {
public static void main(String[] args) throws InterruptedException {
//1:建立一個SparkConf對象
SparkConf conf = new SparkConf()
.setMaster("local[2]")
.setAppName("SparkStreamingFromSocketWithJava");
//2:根據SparkConf對象建立JavaStreamingContext對象,並指定每一個批次處理的時間間隔
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
jssc.sparkContext().setLogLevel("WARN");
//3:註冊一個監聽的ip地址和端口,用來收集數據
JavaReceiverInputDStream<String> lines = jssc.socketTextStream("spark-node01.itheima.com", 9999);
//4:切分每一行記錄
JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
//5:每一個單詞記爲1
JavaPairDStream<String, Integer> pairs = words.mapToPair(word -> new Tuple2<>(word, 1));
//6:對全部的單詞進行聚合輸出
JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey((a, b) -> a + b);
//7:打印數據
wordCounts.print();
//8:開啓流式計算
jssc.start();
//Wait for the computation to terminate 等待計算終止
jssc.awaitTermination();
}
}
注意:因爲使用的是本地模式"local[2]"因此能夠直接在本地運行該程序要指定並行度，如在本地運行設置setMaster("local[2]")，至關於啓動兩個線程，一個給receiver，一個給computer。若是是在集羣中運行，必需要求集羣中可用core數大於1。
Scala語言
object SparkStreamingFromSocketWithScala {
def main(args: Array[String]): Unit = {
//1:建立SparkConf對象,並指定主機名稱設置AppName
val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
//2:根據SparkConf對象建立StreamingContext對象
val ssc = new StreamingContext(conf, Seconds(1))
ssc.sparkContext.setLogLevel("WARN")
//3:讀取socket數據
val lines: ReceiverInputDStream[String] = ssc.socketTextStream("spark-node01.itheima.com",9999)
//4:flatMap
val words: DStream[String] = lines.flatMap(line=>line.split(" "))
//5:map操做
val pairs: DStream[(String, Int)] = words.map(word=>(word,1))
//6:對pairs進行聚合統計
val wordCounts: DStream[(String, Int)] = pairs.reduceByKey((a, b)=>a+b)
//打印結果
wordCounts.print()
//開啓流式計算
ssc.start()
ssc.awaitTermination()
}
}
4.2 Spark Streaming接受socket數據,實現全部批次單詞計數結果累加
在上面的那個案例中存在這樣一個問題：每一個批次的單詞次數都被正確的統計出來，可是結果不能累加！若是將全部批次的結果數據進行累加使用updateStateByKey(func)來更新狀態.
4.2.1 架構圖
4.2.2 代碼實現
4.2.2.1 Java版本
public class SparkStreamingFromSocketTotalWithJava {
public static void main(String[] args) throws InterruptedException {
//1:建立一個SparkConf對象
SparkConf conf = new SparkConf()
.setMaster("local[2]")
.setAppName("SparkStreamingFromSocketTotalWithJava");
//2:根據SparkConf對象建立JavaStreamingContext對象,並指定每一個批次處理的時間間隔
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
jssc.sparkContext().setLogLevel("WARN");
jssc.checkpoint("./ck");
//3:註冊一個監聽的ip地址和端口,用來收集數據
JavaReceiverInputDStream<String> lines = jssc.socketTextStream("spark-node01.itheima.com", 9999);
//4:切分每一行記錄
JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
//5:每一個單詞記爲1
JavaPairDStream<String, Integer> pairs = words.mapToPair(word -> new Tuple2<>(word, 1));
//6:對全部的單詞進行聚合輸出
JavaPairDStream<String, Integer> wordCounts = pairs.updateStateByKey((values,state)->{
Integer updateValue = 0;
if(state.isPresent()){
updateValue = state.get();
}
for (Integer value : values) {
updateValue += value;
}
return Optional.of(updateValue);
});
//7:打印數據
wordCounts.print();
//8:開啓流式計算
jssc.start();
//Wait for the computation to terminate 等待計算終止
jssc.awaitTermination();
}
}
4.2.2.2 Scala版本
object SparkStreamingFromSocketTotalWithScala {
def updateFunction(newValues: Seq[Int], historyValue: Option[Int]): Option[Int] = {
val newCount =historyValue.getOrElse(0)+newValues.sum
Option.apply(newCount)
}
def main(args: Array[String]): Unit = {
//1:建立SparkConf對象
val conf:SparkConf = new SparkConf()
.setMaster("local[2]")
.setAppName("SparkStreamingFromSocketTotalWithScala")
//2:根據SparkConf對象建立StreamingContext對象,並制定時間間隔
val ssc:StreamingContext = new StreamingContext(conf, Seconds(5))
ssc.sparkContext.setLogLevel("WARN")
ssc.checkpoint("./ckscala")
//3:註冊監聽端口,指定主機和端口
val lines: ReceiverInputDStream[String] = ssc.socketTextStream("spark-node01.itheima.com",9999)
//4:切分每一行數據
val words: DStream[String] = lines.flatMap(line=>line.split(" "))
//5:單詞沒出現一次就記爲1
val pairs: DStream[(String, Int)] = words.map(word=>(word,1))
//6:統計單詞出現的次數
val wordCounts: DStream[(String, Int)] = pairs.updateStateByKey(updateFunction)
//7:打印結果
wordCounts.print()
//8:開啓流式計算
ssc.start()
ssc.awaitTermination()
}
}
經過函數updateStateByKey實現。根據key的當前值和key的以前批次值，對key進行更新，返回一個新狀態的DStream.
4.3 SparkStreaming開窗函數reduceByKeyAndWindow
使用Spark Streaming的開窗函數reduceByKeyAndWindow,實現單詞統計計數.
4.3.1 架構圖
4.3.2 代碼實現
4.3.2.1 Java版本
public class SparkStreamingWindowWithJava {
public static void main(String[] args) throws InterruptedException {
//1:建立一個SparkConf對象
SparkConf conf = new SparkConf()
.setMaster("local[4]")
.setAppName("SparkStreamingWindowWithJava");
//2:根據SparkConf對象建立JavaStreamingContext對象,並指定每一個批次處理的時間間隔
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
jssc.sparkContext().setLogLevel("WARN");
//3:註冊一個監聽的ip地址和端口,用來收集數據
JavaReceiverInputDStream<String> lines = jssc.socketTextStream("spark-node01.itheima.com", 9999);
//4:切分每一行記錄
JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
//5:每一個單詞記爲1
JavaPairDStream<String, Integer> pairs = words.mapToPair(word -> new Tuple2<>(word, 1));
//6:進行開窗函數的計算
//reduceByKeyAndWindow 參數說明
//windowDuration:表示window框住的時間長度，如本例5秒切分一次RDD，框10秒，就會保留最近2次切分的RDD
//slideDuration:表示window滑動的時間長度，即每隔多久執行本計算
JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKeyAndWindow((a, b) -> a + b, Durations.seconds(10), Durations.seconds(5));
//7:將結果打印到控制檯
wordCounts.print();
//8:開啓流式計算
jssc.start();
//9:等待終止退出
jssc.awaitTermination();
}
}
4.3.2.2 Scala版本
object SparkStreamingWindowWithScala {
def main(args: Array[String]): Unit = {
//1:建立SparkConf對象
val conf:SparkConf = new SparkConf()
.setMaster("local[2]")
.setAppName("SparkStreamingWindowWithScala")
//2:根據SparkConf對象建立StreamingContext對象,並制定時間間隔
val ssc:StreamingContext = new StreamingContext(conf, Seconds(5))
//3:註冊監聽端口,指定主機和端口
val lines: ReceiverInputDStream[String] = ssc.socketTextStream("spark-node01.itheima.com",9999)
//4:切分每一行
val words: DStream[String] = lines.flatMap(line=>line.split(" "))
//5:單詞出現1次就記爲1(word,1)
val pairs: DStream[(String, Int)] = words.map(word=>(word,1))
//6:對pairs進行開窗統計
//參數說明
//reduceFunc:集合函數
//windowDuration:窗口的寬度,如本例5秒切分一次RDD，框10秒，就會保留最近2次切分的RDD
//slideDuration:表示window滑動的時間長度，即每隔多久執行本計算
val result: DStream[(String, Int)] = pairs.reduceByKeyAndWindow((a:Int, b:Int)=>a+b,Seconds(10),Seconds(5))
//7:打印結果
result.print()
//8:開啓流式計算
ssc.start()
//9:等待終止退出
ssc.awaitTermination()
}
}
現象：Spark Streaming每隔5s計算一次當前在窗口大小爲10s內的數據，而後將結果數據輸出。窗口的寬度和滑動長度都應該是批次處理時間的整數倍.
4.3 實現必定時間內的熱門詞彙搜索
4.3.1 架構圖
4.3.2 代碼實現
4.3.2.1 Java版本
public class SparkStreamingWindowHotWordsWithJava {
public static void main(String[] args) throws InterruptedException {
//1:建立一個SparkConf對象
SparkConf conf = new SparkConf()
.setMaster("local[4]")
.setAppName("SparkStreamingWindowWithJava");
//2:根據SparkConf對象建立JavaStreamingContext對象,並指定每一個批次處理的時間間隔
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
jssc.sparkContext().setLogLevel("WARN");
//3:註冊一個監聽的ip地址和端口,用來收集數據
JavaReceiverInputDStream<String> lines = jssc.socketTextStream("spark-node01.itheima.com", 9999);
//4:切分每一行記錄
JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
//5:每一個單詞記爲1
JavaPairDStream<String, Integer> pairs = words.mapToPair(word -> new Tuple2<>(word, 1));
//6:進行開窗函數的計算
//reduceByKeyAndWindow 參數說明
//windowDuration:表示window框住的時間長度，如本例5秒切分一次RDD，框10秒，就會保留最近2次切分的RDD
//slideDuration:表示window滑動的時間長度，即每隔多久執行本計算
JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKeyAndWindow((a, b) -> a + b, Durations.seconds(10), Durations.seconds(5));
//7:對wordCounts進行排序
JavaPairDStream<String, Integer> finalDStream = wordCounts.transformToPair(wordCount->{
//將統計的每個單詞進行位置替換
JavaPairRDD<Integer, String> reverseWordCountRDD = wordCount.mapToPair(tuple2 -> new Tuple2<Integer, String>(tuple2._2, tuple2._1));
//對反轉以後的rdd進行排序
JavaPairRDD<Integer, String> sortedWordCountRDD = reverseWordCountRDD.sortByKey(false);
//對排序以後的rdd進行再次反轉
JavaPairRDD<String, Integer> reverSortedWordCount = sortedWordCountRDD.mapToPair(tuple2 -> new Tuple2<String, Integer>(tuple2._2, tuple2._1));
//對排序以後的結果取出前三位
List<Tuple2<String, Integer>> take = reverSortedWordCount.take(3);
//打印前三位
for (Tuple2<String, Integer> tuple2 : take) {
System.out.println(tuple2._1+"............."+tuple2._2);
}
return reverSortedWordCount;
});
//7:對wordCounts進行排序
/*JavaPairDStream<String, Integer> finalDStream = wordCounts.transformToPair(new Function<JavaPairRDD<String, Integer>, JavaPairRDD<String, Integer>>() {
@Override