Spark Streaming 編程入門指南

時間 2020-04-14

原文原文鏈接

Spark Streaming 是核心Spark API的擴展，可實現實時數據流的可伸縮，高吞吐量，容錯流處理。能夠從許多數據源（例如Kafka，Flume，Kinesis或TCP sockets）中提取數據，而且可使用複雜的算法處理數據，這些算法用高級函數表示，如map、reduce、join和window。最後，能夠將處理後的數據推送到文件系統，數據庫和實時儀表板。實際上，能夠在數據流上應用Spark的機器學習和圖形處理算法。html

在內部，它的工做方式以下。 Spark Streaming接收實時輸入數據流，並將數據分紅批次，而後由Spark引擎進行處理，以生成批次的最終結果流。java

Spark Streaming提供了一種高級抽象，稱爲離散流或DStream，它表示連續的數據流。DStreams能夠從Kafka、Flume和Kinesis等源的輸入數據流建立，也能夠經過在其餘DStreams上應用高級操做建立。在內部，DStream表示爲RDDs序列。node

1. 瞭解Spark算法

Apache Spark 是一個用於大規模數據處理的統一分析引擎數據庫

特性：apache

快api

將工做負載運行速度提升100倍服務器

Apache Spark使用最新的DAG調度程序，查詢優化器和物理執行引擎，爲批處理數據和流數據提供了高性能。app

易用機器學習

可使用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 }