Spark Streaming 編程入門指南

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 

標籤：大數據

上一篇：已建立好存盤程序，在運行時提示：如正文圖片

下一篇：portlist.ini