Spark Streaming 是核心Spark API的擴充套件，可實現實時資料流的可伸縮，高吞吐量，容錯流處理。可以從許多資料來源（例如Kafka，Flume，Kinesis或TCP sockets）中提取資料，並且可以使用複雜的演算法處理資料，這些演算法用高階函式表示，如map、reduce、join和window。最後，可以將處理後的資料推送到檔案系統，資料庫和實時儀表板。實際上，可以在資料流上應用Spark的機器學習和圖形處理演算法。

在內部，它的工作方式如下。 Spark Streaming接收實時輸入資料流，並將資料分成批次，然後由Spark引擎進行處理，以生成批次的最終結果流。

Spark Streaming提供了一種高階抽象，稱為離散流或DStream，它表示連續的資料流。DStreams可以從Kafka、Flume和Kinesis等源的輸入資料流建立，也可以通過在其他DStreams上應用高階操作建立。在內部，DStream表示為RDDs序列。

1. 瞭解Spark

Apache Spark 是一個用於大規模資料處理的統一分析引擎

特性：

快

將工作負載執行速度提高100倍

Apache Spark使用最新的DAG排程程式，查詢優化器和物理執行引擎，為批處理資料和流資料提供了高效能。

易用

可以使用Java，Scala，Python，R和SQL快速編寫應用程式。

通用

結合SQL、流和複雜的分析

Spark為包括SQL和DataFrames，用於機器學習的MLlib，GraphX和Spark Streaming在內的一堆庫提供支援。您可以在同一應用程式中無縫組合這些庫。

到處執行

Spark可在Hadoop，Apache Mesos，Kubernetes，獨立或雲中執行。它可以訪問各種資料來源。

可以在EC2，Hadoop YARN，Mesos或Kubernetes上使用其獨立叢集模式執行Spark。訪問HDFS，Alluxio，Apache Cassandra，Apache HBase，Apache Hive和數百種其他資料來源中的資料。

2. 入門案例

統計單詞出現的次數，這個例子在Hadoop中用MapReduce也寫過。

JavaStreamingContext是java版的StreamingContext。它是Spark Streaming功能的主要入口點。它提供了從輸入源建立JavaDStream和JavaPairDStream的方法。可以使用context.sparkContext訪問內部的org.apache.spark.api.java.JavaSparkContext。在建立和轉換DStream之後，可以分別使用context.start()和context.stop()啟動和停止流計算。

 1 public static void main(String[] args) throws InterruptedException {
 2     // Create a local StreamingContext with two working thread and batch interval of 1 second
 3     SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
 4     JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));
 5 
 6     // Create a DStream that will connect to hostname:port, like localhost:9999
 7     JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);
 8 
 9     // Split each line into words
10     JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());
11 
12     // Count each word in each batch
13     JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));
14     JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey((i1, i2) -> i1 + i2);
15 
16     // Print the first ten elements of each RDD generated in this DStream to the console
17     wordCounts.print();
18 
19     // Start the computation
20     jssc.start();
21     // Wait for the computation to terminate
22     jssc.awaitTermination();
23 }

3. 基本概念

3.1. Maven依賴

1 <groupId>org.apache.spark</groupId>
2     <artifactId>spark-streaming_2.12</artifactId>
3     <version>2.4.5</version>
4     <scope>provided</scope>
5 </dependency>

為了從其它資料來源獲取資料，需要新增相應的依賴項spark-streaming-xyz_2.12。例如：

1 <dependency>
2     <groupId>org.apache.spark</groupId>
3     <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
4     <version>2.4.5</version>
5 </dependency>

3.2. 初始化StreamingContext

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

我們可以從SparkConf物件中建立一個JavaStreamingContext物件

1 import org.apache.spark.SparkConf;
2 import org.apache.spark.streaming.Duration;
3 import org.apache.spark.streaming.api.java.JavaStreamingContext;
4 
5 SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
6 JavaStreamingContext ssc = new JavaStreamingContext(conf, new Duration(1000)); 

appName 引數是顯示在叢集UI上的你的應用的名字

master 引數是一個Spark、 Mesos 或 YARN 叢集URL，或者也可以是一個特定的字串“local[*]”表示以本地模式執行。實際上，當在叢集上執行時，肯定不希望對在程式中對master進行硬編碼，而希望通過spark-submit啟動應用程式並在其中接收它。然而，對於本地測試，你可以傳“local[*]”來執行Spark Streaming。

還可以從一個已存在的JavaSparkContext中建立一個JavaStreamingContext物件

1 import org.apache.spark.streaming.api.java.*;
2 
3 JavaSparkContext sc = ...   //existing JavaSparkContext
4 JavaStreamingContext ssc = new JavaStreamingContext(sc, Durations.seconds(1));

在定義完context之後，必須做以下事情：

1. 通過建立input DStreams來定義input sources
2. 通過對DStreams應用transformation（轉換）和output（輸出）操作來定義流計算
3. 用streamingContext.start()來開始接收資料並處理它
4. 用streamingContext.awaitTermination()等待處理停止（手動停止或由於任何錯誤）
5. 用streamingContext.stop()可以手動停止

需要記住的點：

• 一旦啟動上下文，就無法設定新的流計算或將其新增到該流計算中
• 上下文一旦停止，就無法重新啟動
• 一個JVM中只能同時啟用一個StreamingContext
• StreamingContext中的stop()也會停止SparkContext。但如果要僅停止StreamingContext的話，設定stop(false)
• 只要在建立下一個StreamingContext之前停止了上一個StreamingContext（不停止SparkContext），就可以將SparkContext重用於建立多個StreamingContext

3.3. DStreams（離散流）

Discretized Stream 或 DStream 是Spark Streaming提供的基本抽象。它表示一個連續的資料流，可以是從源接收的輸入資料流，也可以是通過轉換輸入流生成的已處理資料流。在內部，DStream由一系列連續的RDD表示，這是Spark對不變的分散式資料集的抽象。DStream中的每個RDD都包含來自特定間隔的資料，如下圖所示。 

在DStream上執行的任何操作都轉換為對基礎RDD的操作。例如，最簡單的將一行句子轉換為單詞的例子中，flatMap操作應用於行DStream中的每個RDD，以生成單詞DStream的RDD。如下圖所示：

3.4. Input DStreams 和 Receivers

Input DStream是表示從源接收的輸入資料流。在上圖中，lines是輸入DStream，因為它表示從netcat伺服器接收的資料流。每一個輸入DStream都關聯著一個Receiver物件，該物件從源接收資料並將其儲存在Spark的記憶體中以進行處理。

Spark Streaming提供了兩類內建的streaming源：

• Basic sources ：直接在StreamingContext API中可用的源。例如，檔案系統和socket連線
• Advanced sources ：像Kafka，Flume，Kinesis等這樣的源，可通過額外的程式類獲得 

如果要在流應用程式中並行接收多個數據流，則可以建立多個輸入DStream。這將建立多個Receiver（接收器），這些接收器將同時接收多個數據流。重要的是要記住，必須為Spark Streaming應用程式分配足夠的核心（或執行緒，如果在本地執行），以處理接收到的資料以及執行接收器。

需要記住的點：

• 在本地執行Spark Streaming程式時，請勿使用“ local”或“ local [1]”作為master URL。這兩種方式均意味著僅一個執行緒將用於本地執行任務。如果使用的是基於接收器的輸入DStream（例如套接字，Kafka，Flume等），則將使用單個執行緒來執行接收器，而不會留下任何執行緒來處理接收到的資料。 因此，在本地執行時，請始終使用“ local [n]”作為主URL，其中n>要執行的接收器數 
• 為了將邏輯擴充套件到在叢集上執行，分配給Spark Streaming應用程式的核心數必須大於接收器數。 否則，系統將接收資料，但無法處理它。

Basic Sources 

為了從檔案中讀取資料，可以通過StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass]來建立一個DStream

例如：streamingContext.textFileStream(dataDirectory);

Spark Streaming將監視目錄dataDirectory並處理在該目錄中建立的所有檔案

• 可以監視一個簡單的目錄，例如："hdfs://namenode:8040/logs/2017/*"。在這裡，DStream將由目錄中與模式匹配的所有檔案組成。也就是說：它是目錄的模式，而不是目錄中的檔案。
• 所有檔案必須使用相同的資料格式
• 根據檔案的修改時間而不是建立時間，將其視為時間段的一部分
• 一旦已經被處理後，在當前視窗中對檔案的更改不會導致重新讀取該檔案。即：更新被忽略。

3.5. Transformations on DStreams

對DStreams做轉換，與RDD相似，轉換允許修改輸入DStream中的資料。DStream支援普通Spark RDD上可用的許多轉換。一些常見的方法如下：

其實，這次操作跟Java Stream很像

Window Operations（視窗操作）

Spark Streaming還提供了視窗計算，可以在資料的滑動視窗上應用轉換。下圖說明了此滑動視窗：

如圖所示，每當視窗在源DStream上滑動時，就會對落入視窗內的源RDD進行操作，以生成視窗DStream的RDD。

任何視窗函式所必須的兩個引數：

• 視窗的長度
• 滑到的頻率（或者說時間間隔）

舉個例子，我們來擴充套件前面的示例，假設我們想要每10秒在資料的最後30秒生成一次單詞次數統計。為此，必須在資料的最後30秒內對（word，1）對的DStream對應用reduceByKey操作。

 1 import org.apache.spark.streaming.Durations;
 2 import org.apache.spark.streaming.api.java.JavaDStream;
 3 import org.apache.spark.streaming.api.java.JavaPairDStream;
 4 import scala.Tuple2;
 5 
 6 
 7 JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());
 8 JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));
 9 
10 // Reduce last 30 seconds of data, every 10 seconds
11 JavaPairDStream<String, Integer> windowedWordCounts = pairs.reduceByKeyAndWindow((i1, i2) -> i1 + i2, Durations.seconds(30), Durations.seconds(10));

一些常見的視窗操作如下。所有這些操作均採用上述兩個引數：windowLength和slideInterval

3.6. Output Operations on DStreams

輸出操作允許將DStream的資料輸出到外部系統，例如資料庫或檔案系統。

流式應用程式必須24/7全天候執行，因此必須能夠抵抗與應用程式邏輯無關的故障（例如，系統故障，JVM崩潰等）。為此，Spark Streaming需要將足夠的資訊檢查點指向容錯儲存系統，以便可以從故障中恢復。檢查點有兩種型別的資料。

• 元資料檢查點-將定義流計算的資訊儲存到HDFS等容錯儲存中。這用於從執行流應用程式的驅動程式的節點的故障中恢復。
• 資料檢查點-將生成的RDD儲存到可靠的儲存中 

完整程式碼：

 1 package com.example.demo;
 2 
 3 import org.apache.spark.SparkConf;
 4 import org.apache.spark.streaming.Durations;
 5 import org.apache.spark.streaming.api.java.JavaDStream;
 6 import org.apache.spark.streaming.api.java.JavaPairDStream;
 7 import org.apache.spark.streaming.api.java.JavaStreamingContext;
 8 import scala.Tuple2;
 9 
10 import java.util.Arrays;
11 import java.util.regex.Pattern;
12 
13 /**
14  * @author ChengJianSheng
15  */
16 public class JavaWordCount {
17 
18     private static final Pattern SPACE = Pattern.compile(" ");
19 
20     public static void main(String[] args) {
21         if (args.length < 1) {
22             System.err.println("Usage: JavaWordCount <file>");
23             System.exit(1);
24         }
25 
26         SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("JavaWordCount");
27         JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));
28 
29         JavaDStream<String> lines = jssc.textFileStream(args[0]);
30         JavaDStream<String> words = lines.flatMap(line -> Arrays.asList(SPACE.split(line)).iterator());
31         JavaPairDStream<String, Integer> ones = words.mapToPair(word -> new Tuple2<>(word, 1));
32         JavaPairDStream<String, Integer> counts = ones.reduceByKey((i1, i2) -> i1 + i2);
33         counts.print();
34 
35         /*
36         JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(1));
37         JavaDStream<String> textFileStream = jsc.textFileStream("/data");
38         textFileStream.flatMap(line->Arrays.asList(line.split(" ")).iterator())
39                 .mapToPair(word->new Tuple2<>(word, 1))
40                 .reduceByKey((a,b)->a+b)
41                 .print();
42         jsc.start();
43         */
44     }
45 }

4. Docs

https://spark.apache.org/ 

https://spark.apache.org/docs/latest/streaming-programming-guide.html&n