Spark Streaming 編程入門指南_ZenDei技術網路在線

Spark Streaming 是核心Spark API的擴展，可實現實時數據流的可伸縮，高吞吐量，容錯流處理。可以從許多數據源（例如Kafka，Flume，Kinesis或TCP sockets）中提取數據，並且可以使用複雜的演算法處理數據，這些演算法用高級函數表示，如map、reduce、join和 ...

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之後，必須做以下事情：

通過創建input DStreams來定義input sources
通過對DStreams應用transformation（轉換）和output（輸出）操作來定義流計算
用streamingContext.start()來開始接收數據並處理它
用streamingContext.awaitTermination()等待處理停止（手動停止或由於任何錯誤）
用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上可用的許多轉換。一些常見的方法如下：

map(func)	通過將源DStream的每個元素傳遞給函數func來處理並返回新的DStream
flatMap(func)	與map類似，但是每個輸入項可以映射到0個或多個輸出項
filter(func)	過濾
repartition(numPartitions)	通過創建更多或更少的分區來更改此DStream中的並行度
union(otherStream)	將源DStream和另一個DStream中的元素合併在一起，返回一個新的DStream。相當於SQL中的union
count()	返回元素的個數
reduce(func)	通過使用函數func（接受兩個參數並返回一個）來聚合源DStream的每個RDD中的元素，從而返回一個單元素RDD的新DStream。
countByValue()	在類型為K的元素的DStream上調用時，返回一個新的（K，Long）形式的DStream，其中每個鍵的值是其在源DStream的每個RDD中的頻率。
reduceByKey(func, [numTasks])	在一個（K，V）形式的DStream上調用時，返回一個新的（K，V）DStream，其中使用給定的reduce函數彙總每個鍵的值
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)	通過對源DStream的每個RDD應用RDD-to-RDD函數來返回新的DStream。這可用於在DStream上執行任意RDD操作。
updateStateByKey(func)	返回一個新的“state” DStream

其實，這次操作跟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

window(windowLength, slideInterval)	返回基於源DStream的視窗批處理計算的新DStream
countByWindow(windowLength, slideInterval)	返迴流中元素的滑動視窗數
reduceByWindow(func, windowLength, slideInterval)	對視窗內的數據進行聚合操作
reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])	在（K，V）DStream上調用時，返回新的（K，V）DStream，其中使用給定的reduce函數func在滑動視窗中的批處理上彙總每個鍵的值
reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])
countByValueAndWindow(windowLength, slideInterval, [numTasks])

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