當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

SparkStreaming编程

發(fā)布時間：2024/3/13 编程问答 55 豆豆

生活随笔收集整理的這篇文章主要介紹了 SparkStreaming编程小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.

0. SparkStreaming

流式計算簡介
SparkStreaming實時處理入門案例
SparkStreaming和HDFS整合
SparkStreaming與Kafka整合
SparkStreaming常見transformation算子
SparkStreaming高可用及其優(yōu)化建議

1. 流式計算簡介

1.1. 流式計算

? 如何去理解流式計算，最形象的例子，就是小明的往水池中放(入)水又放(出)水的案例。流式計算就像水流一樣，數(shù)據(jù)連綿不斷的產生，并被快速處理，所以流式計算擁有如下一些特點：

數(shù)據(jù)是無界的(unbounded)
數(shù)據(jù)是動態(tài)的
計算速度是非常快的
計算不止一次
計算不能終止

? 反過來看看一下離線計算有哪些特點：

數(shù)據(jù)是有界的(Bounded)
數(shù)據(jù)靜態(tài)的
計算速度通常較慢
計算只執(zhí)行一次
計算終會終止

1.2. 常見的離線和流式計算框架

常見的離線計算框架
mapreduce
spark-core
flink-dataset
常見的流式計算框架
storm(jstorm)

第一代的流式處理框架，每生成一條記錄，提交一次作業(yè)。實時流處理，延遲低。
spark-streaming

第二代的流式處理框架，短時間內生成mirco-batch，提交一次作業(yè)。準實時，延遲略高，秒級或者亞秒級延遲。
flink-datastream(blink)

第三代的流式處理框架，每生成一條記錄，提交一次作業(yè)。實時，延遲低。

1.3. SparkStreaming簡介

? SparkStreaming，和SparkSQL一樣，也是Spark生態(tài)棧中非常重要的一個模塊，主要是用來進行流式計算的框架。流式計算框架，從計算的延遲上面，又可以分為純實時流式計算和準實時流式計算，SparkStreaming是屬于的準實時計算框架。

? 所謂純實時的計算，指的是來一條記錄(event事件)，啟動一次計算的作業(yè)；離線計算，指的是每次計算一個非常大的一批(比如幾百G，好幾個T)數(shù)據(jù)；準實時呢，介于純實時和離線計算之間的一種計算方式。顯然不是每一條記錄就計算一次，顯然比起離線計算數(shù)據(jù)量小的多，怎么表示？Micro-batch（微小的批次）。

? SparkStreaming是SparkCore的api的一種擴展，使用DStream(discretized stream or DStream)作為數(shù)據(jù)模型，基于內存處理連續(xù)的數(shù)據(jù)流，本質上還是RDD的基于內存的計算。

? DStream，本質上是RDD的序列。SparkStreaming的處理流程可以歸納為下圖：

1.4. SparkStreaming基本工作原理

? 接收實時輸入數(shù)據(jù)流，然后將數(shù)據(jù)拆分成多個batch，比如每收集1秒的數(shù)據(jù)封裝為一個batch，然后將每個batch交給Spark的計算引擎進行處理，最后會生產出一個結果數(shù)據(jù)流，其中的數(shù)據(jù)，也是由一個一個的batch所組成的。

? Spark Streaming提供了一種高級的抽象，叫做DStream，英文全稱為Discretized Stream，中文翻譯為“離散流”，它代表了一個持續(xù)不斷的數(shù)據(jù)流。DStream可以通過輸入數(shù)據(jù)源來創(chuàng)建，比如Kafka、Flume、ZMQ和Kinesis；也可以通過對其他DStream應用高階函數(shù)來創(chuàng)建，比如map、reduce、join、window。
? DStream的內部，其實一系列持續(xù)不斷產生的RDD。RDD是Spark Core的核心抽象，即，分布式式彈性數(shù)據(jù)集。DStream中的每個RDD都包含了一個時間段內的數(shù)據(jù)。

? 對DStream應用的算子，比如map，其實在底層會被翻譯為對DStream中每個RDD的操作。比如對一個DStream執(zhí)行一個map操作，會產生一個新的DStream。但是，在底層，其實其原理為，對輸入DStream中每個時間段的RDD，都應用一遍map操作，然后生成的新的RDD，即作為新的DStream中的那個時間段的一個RDD。底層的RDD的transformation操作。
? 還是由Spark Core的計算引擎來實現(xiàn)的。Spark Streaming對Spark Core進行了一層封裝，隱藏了細節(jié)，然后對開發(fā)人員提供了方便易用的高層次的API。

1.5. Storm V.S. SparkStreaming V.S. Flink

三者對比

storm和flink簡介

storm: storm.apache.org

flink: flink.apache.org

1.6. 如何選擇一款合適的流式處理框架

對于Storm來說：
1、建議在需要純實時，不能忍受1秒以上延遲的場景下使用，比如實時計算系統(tǒng)，要求純實時進行交易和分析時。
2、在實時計算的功能中，要求可靠的事務機制和可靠性機制，即數(shù)據(jù)的處理完全精準，一條也不能多，一條也不能少，也可以考慮使用Storm，但是Spark Streaming也可以保證數(shù)據(jù)的不丟失。
3、如果我們需要考慮針對高峰低峰時間段，動態(tài)調整實時計算程序的并行度，以最大限度利用集群資源（通常是在小型公司，集群資源緊張的情況），我們也可以考慮用Storm
對于Spark Streaming來說：
1、不滿足上述3點要求的話，我們可以考慮使用Spark Streaming來進行實時計算。
2、考慮使用Spark Streaming最主要的一個因素，應該是針對整個項目進行宏觀的考慮，即，如果一個項目除了實時計算之外，還包括了離線批處理、交互式查詢、圖計算和MLIB機器學習等業(yè)務功能，而且實時計算中，可能還會牽扯到高延遲批處理、交互式查詢等功能，那么就應該首選Spark生態(tài)，用Spark Core開發(fā)離線批處理，用Spark SQL開發(fā)交互式查詢，用Spark Streaming開發(fā)實時計算，三者可以無縫整合，給系統(tǒng)提供非常高的可擴展性。
對于Flink來說：
支持高吞吐、低延遲、高性能的流處理
支持帶有事件時間的窗口（Window）操作
支持有狀態(tài)計算的Exactly-once語義
支持高度靈活的窗口（Window）操作，支持基于time、count、session，以及data-driven的窗口操作
支持具有Backpressure功能的持續(xù)流模型
支持基于輕量級分布式快照（Snapshot）實現(xiàn)的容錯
一個運行時同時支持Batch on Streaming處理和Streaming處理
Flink在JVM內部實現(xiàn)了自己的內存管理
支持迭代計算
支持程序自動優(yōu)化：避免特定情況下Shuffle、排序等昂貴操作，中間結果有必要進行緩存

2. SparkStreaming實時處理入門案例

2.1. 創(chuàng)建項目模塊

指定maven左表

執(zhí)行存儲位置

導入maven依賴

<dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.11</artifactId><version>2.2.2</version> </dependency>

完整的pom文件

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><parent><artifactId>spark-parent-1903</artifactId><groupId>com.desheng.parent</groupId><version>1.0-SNAPSHOT</version></parent><modelVersion>4.0.0</modelVersion><groupId>com.desheng.bigdata</groupId><artifactId>spark-streaming</artifactId><version>1.0-SNAPSHOT</version><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target></properties><dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.11</artifactId></dependency></dependencies> </project>

2.2. 編碼

2.2.1. 入口類StreamingContext

? SparkStreaming中的入口類，稱之為StreamingContext,但是底層還是得需要依賴SparkContext。

object _01SparkStreamingWordCountOps {def main(args: Array[String]): Unit = {/*StreamingContext的初始化，需要至少兩個參數(shù)，SparkConf和BatchDurationSparkConf不用多說batchDuration：提交兩次作業(yè)之間的時間間隔，每次會提交一個DStream，將數(shù)據(jù)轉化batch--->RDD所以說：sparkStreaming的計算，就是每隔多長時間計算一次數(shù)據(jù)*/val conf = new SparkConf().setAppName("SparkStreamingWordCount").setMaster("local[*]")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)//業(yè)務//為了執(zhí)行的流式計算，必須要調用start來啟動ssc.start()//為了不至于start啟動程序結束，必須要調用awaitTermination方法等待程序業(yè)務完成之后調用stop方法結束程序，或者異常ssc.awaitTermination()} }

2.2.2. 業(yè)務編寫

object _01SparkStreamingWordCountOps {def main(args: Array[String]): Unit = {if(args == null || args.length < 2) {println("""|Usage: <hostname> <port>""".stripMargin)System.exit(-1)}val Array(hostname, port) = args/*StreamingContext的初始化，需要至少兩個參數(shù)，SparkConf和BatchDurationSparkConf不用多說batchDuration：提交兩次作業(yè)之間的時間間隔，每次會提交一個DStream，將數(shù)據(jù)轉化batch--->RDD所以說：sparkStreaming的計算，就是每隔多長時間計算一次數(shù)據(jù)*/val conf = new SparkConf().setAppName("SparkStreamingWordCount").setMaster("local[*]")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)//接入數(shù)據(jù)val lines:ReceiverInputDStream[String] = ssc.socketTextStream(hostname, port.toInt)// lines.print()val retDStream:DStream[(String, Int)] = lines.flatMap(_.split("\\s+")).map((_, 1)).reduceByKey(_+_)retDStream.print()//為了執(zhí)行的流式計算，必須要調用start來啟動ssc.start()//為了不至于start啟動程序結束，必須要調用awaitTermination方法等待程序業(yè)務完成之后調用stop方法結束程序，或者異常ssc.awaitTermination()} }

2.2.3. 使用netcat進行測試

2.3. StreamingContext和Receiver說明

2.3.1. StreamingContext

? StreamingContext是程序的入口類，用于創(chuàng)建DStream，維護SparkStreaming程序的聲明周期。

關于local說明

當我們將上述程序中的master由local[*],修改為local的時候，程序業(yè)務不變，發(fā)生只能接收數(shù)據(jù)，無法處理數(shù)據(jù)。

local[*]和local的區(qū)別，后者只為當前程序提供一個線程來處理，前者提供可用的所有的cpu的core來處理，當前情況下為2或者4。所以我們推測，當前程序無法處理數(shù)據(jù)的原因，只能是擁有cpu-core或者線程個數(shù)造成的。同時還可以推到出來的是，SparkStreaming在當下案例中，優(yōu)先使用線程資源來接收數(shù)據(jù)，其次才是對數(shù)據(jù)的處理，接收數(shù)據(jù)的對象就是Receiver。所以，以后注意，如果讀取數(shù)據(jù)的時候有receiver，程序的線程個數(shù)至少為2。

start

start方法是用來啟動當前sparkStreaming應用的，所以，是不能在ssc.start()之后再添加任何業(yè)務邏輯，否則，涼涼！

- awaitTermination

2.3.2. Receiver

? Receiver，顧名思義，就是數(shù)據(jù)的接收者，這里把資源分成了兩部分，一部分用來接收數(shù)據(jù)，一部分用來處理數(shù)據(jù)。Receiver接收到的數(shù)據(jù)，說白了就是一個個的batch數(shù)據(jù)，是RDD，存儲在Executor內存。Receiver就是Executor內存中的一部分。

? 不是所有的streaming作業(yè)都需要有Receiver。

? 通過下圖，來闡述基于Receiver的程序執(zhí)行的流程

3. SparkStreaming和HDFS整合

3.1. 說明

? SparkStreaming監(jiān)聽hdfs的某一個目錄，目錄下的新增文件，做實時處理。這種方式在特定情況下還是挺多的。需要使用的api為：ssc.fileStream()。

? 監(jiān)聽的文件，必須要從另一個相匹配的目錄移動到其它目錄。

監(jiān)聽本地

無法讀取手動拷貝，或者剪切到指定目錄下的文件，只能讀取通過流寫入的文件。
監(jiān)聽hdfs

有的操作系統(tǒng)和監(jiān)聽本地是一樣。

正常情況下，我們可以讀取到通過put上傳的文件，還可以讀取通過cp拷貝的文件，但是讀取不了mv移動的文件。

讀取文件的這種方式，沒有額外的Receiver消耗線程資源，所以可以指定master為local

3.2. 編碼

object _01SparkStreamingHDFS {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("SparkStreamingHDFS").setMaster("local")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)//讀取local中數(shù)據(jù) --->需要通過流的方式寫入 // val lines = ssc.textFileStream("file:///E:/data/monitored")//hdfsval lines = ssc.textFileStream("hdfs://bigdata01:9000/data/spark")lines.print()ssc.start()ssc.awaitTermination()} }

4. SparkStreaming與Kafka整合（*）

4.1. 整合簡述

? kafka是做消息的緩存，數(shù)據(jù)和業(yè)務隔離操作的消息隊列，而sparkstreaming是一款準實時流式計算框架，所以二者的整合，是大勢所趨。

? 二者的整合，有主要的兩大版本。

? 在spark-stremaing-kafka-0-8的版本中又分為了兩種方式：receiver的方式和direct的方式來讀取kafka中的數(shù)據(jù)，主要區(qū)別就是是否依賴zookeeper來管理offset信息，以及是否擁有receiver。

4.2. spark-stremaing-kafka-0-8

api地址：

? http://spark.apache.org/docs/2.2.2/streaming-kafka-0-8-integration.html

導入依賴

<dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-8_2.11</artifactId><version>2.2.2</version> </dependency>

入口類便是KafkaUtils

4.2.1. Receiver的方式

編程
/*** 使用kafka的receiver-api讀取數(shù)據(jù)*/ object _02SparkStreamingWithKafkaReceiverOps {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("SparkStreamingWithKafkaReceiver").setMaster("local[*]")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)val kafkaParams = Map[String, String]("zookeeper.connect" -> "bigdata01:2181,bigdata02:2181,bigdata03:2181/kafka","group.id" -> "g_1903_1","zookeeper.connection.timeout.ms" -> "10000")val topics = Map[String, Int]("spark" -> 3)val messages:ReceiverInputDStream[(String, String)] = KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics,StorageLevel.MEMORY_AND_DISK_SER_2)messages.print()ssc.start()ssc.awaitTermination()} }
說明
這種方式使用Receiver來獲取數(shù)據(jù)。Receiver是使用Kafka的高層次Consumer API來實現(xiàn)的。receiver從Kafka中獲取的數(shù)據(jù)都是存儲在Spark Executor的內存中的，然后Spark Streaming啟動的job會去處理那些數(shù)據(jù)。然而，在默認的配置下，這種方式可能會因為底層的失敗而丟失數(shù)據(jù)。如果要啟用高可靠機制，讓數(shù)據(jù)零丟失，就必須啟用Spark Streaming的預寫日志機制（Write Ahead Log，WAL）。該機制會同步地將接收到的Kafka數(shù)據(jù)寫入分布式文件系統(tǒng)（比如HDFS）上的預寫日志中。所以，即使底層節(jié)點出現(xiàn)了失敗，也可以使用預寫日志中的數(shù)據(jù)進行恢復。
需要注意的地方
Kafka的topic分區(qū)和Spark Streaming中生成的RDD分區(qū)沒有關系。在KafkaUtils.createStream中增加分區(qū)數(shù)量只會增加單個receiver的線程數(shù)，不會增加Spark的并行度
可以創(chuàng)建多個的Kafka的輸入DStream，使用不同的group和topic，使用多個receiver并行接收數(shù)據(jù)。
如果啟用了HDFS等有容錯的存儲系統(tǒng)，并且啟用了寫入日志，則接收到的數(shù)據(jù)已經(jīng)被復制到日志中。因此，輸入流的存儲級別設置StorageLevel.MEMORY_AND_DISK_SER（即使用KafkaUtils.createStream（…，StorageLevel.MEMORY_AND_DISK_SER））的存儲級別。
數(shù)據(jù)會丟失原因

4.2.2. Direct的方式

編碼
//基于direct方式整合kafka object _03SparkStreamingWithKafkaDirectOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("SparkStreamingWithKafkaDirect").setMaster("local[*]")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)val kafkaParams = Map[String, String]("bootstrap.servers" -> "bigdata01:9092,bigdata02:9092,bigdata03:9092","group.id" -> "g_1903_2","auto.offset.reset" -> "largest")val topics = "spark".split(",").toSetval messages: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)messages.foreachRDD((rdd, bTime) => {if(!rdd.isEmpty()) {val offsetRDD = rdd.asInstanceOf[HasOffsetRanges]val offsetRanges = offsetRDD.offsetRangesfor(offsetRange <- offsetRanges) {val topic = offsetRange.topicval partition = offsetRange.partitionval fromOffset = offsetRange.fromOffsetval untilOffset = offsetRange.untilOffsetprintln(s"topic:${topic}\tpartition:${partition}\tstart:${fromOffset}\tend:${untilOffset}")}rdd.count()}})ssc.start()ssc.awaitTermination()} }
說明
簡化的并行性：不需要創(chuàng)建多個輸入Kafka流并將其合并。使用directStream，Spark Streaming將創(chuàng)建與使用Kafka分區(qū)一樣多的RDD分區(qū)，這些分區(qū)將全部從Kafka并行讀取數(shù)據(jù)。所以在Kafka和RDD分區(qū)之間有一對一的映射關系。
效率：在第一種方法中實現(xiàn)零數(shù)據(jù)丟失需要將數(shù)據(jù)存儲在預寫日志中，這會進一步復制數(shù)據(jù)。這實際
上是效率低下的，因為數(shù)據(jù)被有效地復制了兩次:一次是Kafka，另一次是由預先寫入日志（Write
Ahead Log）復制。這個第二種方法消除了這個問題，因為沒有接收器，因此不需要預先寫入日志。
只要Kafka數(shù)據(jù)保留時間足夠長。
正好一次（Exactly-once）的語義：第一種方法使用Kafka的高級API來在Zookeeper中存儲消耗的偏移量。傳統(tǒng)上這是從Kafka消費數(shù)據(jù)的方式。雖然這種方法（結合提前寫入日志）可以確保零數(shù)據(jù)丟失（即至少一次語義），但是在某些失敗情況下，有一些記錄可能會消費兩次。發(fā)生這種情況是因為Spark Streaming可靠接收到的數(shù)據(jù)與Zookeeper跟蹤的偏移之間的不一致。因此，在第二種方法中，我們使用不使用Zookeeper的簡單Kafka API。在其檢查點內，Spark Streaming跟蹤偏移量。這消除了Spark Streaming和Zookeeper/Kafka之間的不一致，因此Spark Streaming每次記錄都會在發(fā)生故障的情況下有效地收到一次。為了實現(xiàn)輸出結果的一次語義，將數(shù)據(jù)保存到外部數(shù)據(jù)存儲區(qū)的輸出操作必須是冪等的，或者是保存結果和偏移量的原子事務。

冪等說明：多次操作結果都一樣，把這種操作稱之為冪等操作，比如數(shù)據(jù)庫的delete操作，或者：
INSERT INTO USER (id, NAME, age) VALUES(3, 'wangwu1', 16) ON DUPLICATE KEY UPDATE age=16, NAME='wangwu1'
offset的問題
/*offset的checkpoint(檢查點)把需要管理的相關數(shù)據(jù)保存在某一個目錄下面，后續(xù)的時候直接從該目錄中讀取即可，在此處就是保存offset數(shù)據(jù)*/ object _04CheckpointWithKafkaDirectOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("CheckpointWithKafkaDirectOps").setMaster("local")val duration = Seconds(2)val checkpoint = "file:///E:/data/monitored/chk"def createFunc():StreamingContext = {val ssc = new StreamingContext(conf, duration)ssc.checkpoint(checkpoint)val kafkaParams = Map[String, String]("bootstrap.servers" -> "bigdata01:9092,bigdata02:9092,bigdata03:9092","group.id" -> "g_1903_2","auto.offset.reset" -> "smallest")val topics = "spark".split(",").toSetval messages: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)messages.foreachRDD((rdd, bTime) => {if(!rdd.isEmpty()) {println("num: " + rdd.getNumPartitions)val offsetRDD = rdd.asInstanceOf[HasOffsetRanges]val offsetRanges = offsetRDD.offsetRangesfor(offsetRange <- offsetRanges) {val topic = offsetRange.topicval partition = offsetRange.partitionval fromOffset = offsetRange.fromOffsetval untilOffset = offsetRange.untilOffsetprintln(s"topic:${topic}\tpartition:${partition}\tstart:${fromOffset}\tend:${untilOffset}")}rdd.count()}})ssc}//創(chuàng)建或者恢復出來一個StreamingContextval ssc = StreamingContext.getOrCreate(checkpoint, createFunc)ssc.start()ssc.awaitTermination()} }
說明，這種方式，雖然能夠解決offset跟蹤問題，但是會在checkpoint目錄下面產生大量的小文件，并且操作進行磁盤的IO操作，性能相對較差，那我們如何管理offset偏移量呢？

常見的管理offset偏移量的方式有如下：zookeeper、redis、hbase、mysql、elasticsearch、kafka，zookeeper也不建議使用，zookeeper太重要了，zk負載過高，容易出故障。
zookeeper管理offset
/*基于direct方式整合kafka使用zk手動管理offset*/object _05KafkaDirectWithZKOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("SparkStreamingWithKafkaDirect").setMaster("local")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)val kafkaParams = Map[String, String]("bootstrap.servers" -> "bigdata01:9092,bigdata02:9092,bigdata03:9092","group.id" -> "g_1903_2","auto.offset.reset" -> "smallest")val topics = "spark".split(",").toSetval messages: InputDStream[(String, String)] = createMsg(ssc, kafkaParams, topics)messages.foreachRDD((rdd, bTime) => {if(!rdd.isEmpty()) {println("-------------------------------------------")println(s"Time: $bTime")println("###########count: " + rdd.count())storeOffsets(rdd.asInstanceOf[HasOffsetRanges].offsetRanges, kafkaParams("group.id"))println("-------------------------------------------")}})ssc.start()ssc.awaitTermination()}/*從zk中讀取手動保存offset信息，然后從kafka指定offset位置開始讀取數(shù)據(jù)，當然如果沒有讀取到offset信息，那么從最開始或者從最新的位置開始讀取信息*/def createMsg(ssc: StreamingContext, kafkaParams: Map[String, String], topics: Set[String]): InputDStream[(String, String)] = {//從zk中讀取offsetval fromOffsets: Map[TopicAndPartition, Long] = getFromOffsets(topics, kafkaParams("group.id"))var messages: InputDStream[(String, String)] = nullif(fromOffsets.isEmpty) {//沒有讀到offsetmessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)} else {//讀取到了offset，從指定位置開始讀取val messageHandler = (msgHandler:MessageAndMetadata[String, String]) => (msgHandler.key(), msgHandler.message())messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc,kafkaParams, fromOffsets, messageHandler)}messages}/*從zk中讀取offset信息首先，定義offset交互的信息，比如數(shù)據(jù)的存放位置，存放格式官方的操作：/kafka/consumers/${group.id}/offsets/${topic}/${partition} -->data為offset存放位置&數(shù)據(jù)：/kafka/consumers/offsets/${topic}/${group.id}/${partition} -->data為offset其次，用什么和zk中交互zookeeper原生的apicuratorFramework(選擇)*/def getFromOffsets(topics:Set[String], group:String):Map[TopicAndPartition, Long] = {val offsets = mutable.Map[TopicAndPartition, Long]()for (topic <- topics) {val path = s"${topic}/${group}"//判斷當前路徑是否存在checkExists(path)for(partition <- JavaConversions.asScalaBuffer(client.getChildren.forPath(path))) {val fullPath = s"${path}/${partition}"val offset = new String(client.getData.forPath(fullPath)).toLongoffsets.put(TopicAndPartition(topic, partition.toInt), offset)}}offsets.toMap}def storeOffsets(offsetRanges: Array[OffsetRange], group:String) = {for(offsetRange <- offsetRanges) {val topic = offsetRange.topicval partition = offsetRange.partitionval offset = offsetRange.untilOffsetval path = s"${topic}/${group}/${partition}"checkExists(path)client.setData().forPath(path, offset.toString.getBytes)}}def checkExists(path:String): Unit = {if(client.checkExists().forPath(path) == null) {//路徑不能存在client.create().creatingParentsIfNeeded().forPath(path)}}val client = {val client = CuratorFrameworkFactory.builder().connectString("bigdata01:2181,bigdata02:2181,bigdata03:2181").retryPolicy(new ExponentialBackoffRetry(1000, 3)).namespace("kafka/consumers/offsets").build()client.start()client} }
redis管理offset

導入redis的maven依賴
<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>2.9.0</version> </dependency>
提供redis的操作工具類(在spark-common模塊中處理)
//入口類：Jedis -->代表的是redis的一個客戶端連接，相當于Connection public class JedisUtil {static JedisPool pool;static {JedisPoolConfig config = new JedisPoolConfig();String host = "bigdata01";int port = 6379;pool = new JedisPool(config, host, port);}public static Jedis getJedis() {return pool.getResource();}public static void release(Jedis jedis) {jedis.close();} }
在spark-streaming模塊中引入
<dependency><groupId>com.desheng.bigdata</groupId><artifactId>spark-common</artifactId><version>1.0-SNAPSHOT</version> </dependency>
代碼實現(xiàn)
/*基于direct方式整合kafka使用redis手動管理offset*/object _06KafkaDirectWithRedisOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("KafkaDirectWithRedis").setMaster("local")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)val kafkaParams = Map[String, String]("bootstrap.servers" -> "bigdata01:9092,bigdata02:9092,bigdata03:9092","group.id" -> "g_1903_2","auto.offset.reset" -> "smallest")val topics = "spark".split(",").toSetval messages: InputDStream[(String, String)] = createMsg(ssc, kafkaParams, topics)messages.foreachRDD((rdd, bTime) => {if(!rdd.isEmpty()) {println("-------------------------------------------")println(s"Time: $bTime")println("###########count: " + rdd.count())storeOffsets(rdd.asInstanceOf[HasOffsetRanges].offsetRanges, kafkaParams("group.id"))println("-------------------------------------------")}})ssc.start()ssc.awaitTermination()}/*從zk中讀取手動保存offset信息，然后從kafka指定offset位置開始讀取數(shù)據(jù)，當然如果沒有讀取到offset信息，那么從最開始或者從最新的位置開始讀取信息*/def createMsg(ssc: StreamingContext, kafkaParams: Map[String, String], topics: Set[String]): InputDStream[(String, String)] = {//從zk中讀取offsetval fromOffsets: Map[TopicAndPartition, Long] = getFromOffsets(topics)var messages: InputDStream[(String, String)] = nullif(fromOffsets.isEmpty) {//沒有讀到offsetmessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)} else {//讀取到了offset，從指定位置開始讀取val messageHandler = (msgHandler:MessageAndMetadata[String, String]) => (msgHandler.key(), msgHandler.message())messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc,kafkaParams, fromOffsets, messageHandler)}messages}/*從redis中讀取offset信息K-V首先，定義offset交互的信息，比如數(shù)據(jù)的存放位置，存放格式topic partition offset group五種數(shù)據(jù)結構：stringtopic|group|partition offsetlistabchashkey field-value ...topic group|partition offsetgroup topic|partition offsetsetzset其次，用什么和redis中交互*/def getFromOffsets(topics:Set[String]):Map[TopicAndPartition, Long] = {val offsets = mutable.Map[TopicAndPartition, Long]()val jedis = JedisUtil.getJedisfor (topic <- topics) {val gpo = jedis.hgetAll(topic)for((gp, offset) <- gpo) {val partition = gp.substring(gp.indexOf("|") + 1).toIntoffsets.put(TopicAndPartition(topic, partition), offset.toLong)}}JedisUtil.release(jedis)offsets.toMap}def storeOffsets(offsetRanges: Array[OffsetRange], group:String) = {val jedis = JedisUtil.getJedisfor(offsetRange <- offsetRanges) {val topic = offsetRange.topicval partition = offsetRange.partitionval offset = offsetRange.untilOffsetval gp = s"${group}|${partition}"jedis.hset(topic, gp, offset.toString)}JedisUtil.release(jedis)} }
數(shù)據(jù)結構

基于HBase進行管理offset

引入hbase的pom依賴
<dependency><groupId>org.apache.hbase</groupId><artifactId>hbase-server</artifactId><version>1.1.5</version> </dependency>
并在spark-common中完成依賴繼承

在common模塊中完成hbaseconnection工具類的構造
package com.desheng.bigdata.util;import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.filter.BinaryComparator; import org.apache.hadoop.hbase.filter.CompareFilter; import org.apache.hadoop.hbase.filter.Filter; import org.apache.hadoop.hbase.filter.RowFilter;import java.io.IOException; import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map;//hbase connection的工具類，主要提供和回收Connection對象 public class HBaseConnectionPool {private static LinkedList<Connection> pool = new LinkedList<Connection>();//初始化static {try {Configuration conf = HBaseConfiguration.create();for (int i = 0; i < 5; i++) {pool.push(ConnectionFactory.createConnection(conf));}} catch (IOException e) {e.printStackTrace();}}public static Connection getConnection() {while(pool.isEmpty()) {try {System.out.println("Connection pool為空，稍后再試~~");Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}}return pool.poll();}public static void release(Connection connection) {pool.push(connection);}public static String getColValue(Connection connection, TableName tableName, byte[] rk, byte[] cf, byte[] col) {try {Table table = connection.getTable(tableName);Result result = table.get(new Get(rk));return new String(result.getValue(cf, col));} catch (IOException e) {e.printStackTrace();}return null;}/*rowkeytopic-group@return col-value*/public static Map<Integer, Long> getColValue(Connection connection, TableName tableName, byte[] rk, byte[] cf) {Map<Integer, Long> partition2Offset = new HashMap<>();try {Table table = connection.getTable(tableName);Scan scan = new Scan();Filter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new BinaryComparator(rk));scan.setFilter(filter);ResultScanner results = table.getScanner(scan);for(Result result : results) {List<Cell> cells = result.listCells();for(Cell cell : cells) {//colbyte[] qArr = cell.getQualifierArray();int qLen = cell.getQualifierLength();int qOffset = cell.getQualifierOffset();//valuebyte[] vArr = cell.getValueArray();int vLen = cell.getValueLength();int vOffset = cell.getValueOffset();int partition = Integer.valueOf(new String(qArr, qOffset, qLen));long value = Long.valueOf(new String(vArr, vOffset, vLen));partition2Offset.put(partition, value);}}table.close();} catch (IOException e) {e.printStackTrace();}return partition2Offset;}public static void set(Connection connection, TableName tableName, byte[] rk, byte[] cf, byte[] col, byte[] value) {try {Table table = connection.getTable(tableName);Put put = new Put(rk);put.addColumn(cf, col, value);table.put(put);table.close();} catch (IOException e) {e.printStackTrace();}} }
需要涉及hbase-site.xml配置文件
<?xml version="1.0"?> <?xml-stylesheet type="text/xsl" href="configuration.xsl"?><configuration><property><name>hbase.rootdir</name><value>hdfs://ns1/hbase</value></property><property><name>hbase.cluster.distributed</name><value>true</value></property><property><name>hbase.zookeeper.quorum</name><value>bigdata01,bigdata02,bigdata03</value></property> <property><name>hbase.regionserver.wal.codec</name><value>org.apache.hadoop.hbase.regionserver.wal.IndexedWALEditCodec</value></property> </configuration>
offset消費的編碼
package com.desheng.bigdata.p2import com.desheng.bigdata.util.HBaseConnectionPool import kafka.common.TopicAndPartition import kafka.message.MessageAndMetadata import kafka.serializer.StringDecoder import org.apache.hadoop.hbase.TableName import org.apache.spark.SparkConf import org.apache.spark.streaming.dstream.InputDStream import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils, OffsetRange} import org.apache.spark.streaming.{Seconds, StreamingContext}import scala.collection.JavaConversions._ import scala.collection.mutable/*使用hbase來來手動管理offset信息保證數(shù)據(jù)被依次消費1.手動從hbase中讀取上一次消費的offset信息有：從指定的offset位置開始消費無：從offset為0或者最小的位置開始消費2.使用指定offset從kafka中拉取數(shù)據(jù)3.拉取到數(shù)據(jù)之后進行業(yè)務處理4.offset需要更新會指定的hbasecreate 'hadoop-topic-offset', 'cf'rowkeytopic-groupcolumnpartitionoffsetConnection*/ object _07SparkStreamingKafkaHBaseOps {def main(args: Array[String]): Unit = {if(args == null || args.length < 3) {println("""|Usage: <broker.list> <groupId> <topicStr>""".stripMargin)System.exit(-1)}//topicStr就是多個topic的列表，比如：a,b,cval Array(brokerList, groupId, topicStr) = argsval topics = topicStr.split(",").toSetval conf = new SparkConf().setAppName("_07SparkStreamingKafkaHBaseOps").setMaster("local")val batchInterval = Seconds(2)val kafkaParams:Map[String, String] = Map[String, String]("metadata.broker.list" -> brokerList,"group.id" -> groupId,"auto.offset.reset" -> "smallest")val ssc = new StreamingContext(conf, batchInterval)//接入kafka中的數(shù)據(jù)(K, V),k就是一條kafka-record中的key的類型（一般都為null），V就是record中value的類型val input:InputDStream[(String, String)] = createMsg(ssc, kafkaParams, topics)//進行各種業(yè)務邏輯操作//遍歷dstream中每一個rddinput.foreachRDD((rdd, time) => {if(!rdd.isEmpty()) {println("-------------------------------------------")println(s"Time: $time")println("######################rdd's count: " + rdd.count())//更新偏移量storeOffsets(rdd.asInstanceOf[HasOffsetRanges].offsetRanges, kafkaParams("group.id"))println("-------------------------------------------")}})ssc.start()ssc.awaitTermination()}/*** 保存偏移量*/def storeOffsets(offsetRanges: Array[OffsetRange], group:String): Unit = {val connection = HBaseConnectionPool.getConnectionval tableName = TableName.valueOf("hadoop-topic-offset")val cf = "cf".getBytes()for(offsetRange <- offsetRanges) {val rk = s"${offsetRange.topic}-${group}".getBytes()val partition = offsetRange.partitionval offset = offsetRange.untilOffset//將結果保存到hbase中HBaseConnectionPool.set(connection, tableName, rk, cf, (partition + "").getBytes(), (offset + "").getBytes())}HBaseConnectionPool.release(connection)}def createMsg(ssc:StreamingContext, kafkaParams:Map[String, String], topics:Set[String]):InputDStream[(String, String)] = {//從zk中讀取offsetval offsets:Map[TopicAndPartition, Long] = getOffsets(topics, kafkaParams("group.id"))var messages:InputDStream[(String, String)] = nullif(offsets.isEmpty) {//空的，沒有讀取到offset，從0開始讀取messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc,kafkaParams, topics)} else {//有，從指定的offset位置開始讀取offsetval messageHandler = (msgH: MessageAndMetadata[String, String]) => (msgH.key(), msgH.message())messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc,kafkaParams, offsets,messageHandler)}messages}/*** create 'hadoop-topic-offset', 'cf'rowkeytopic-group* @return offsets*/def getOffsets(topics:Set[String], group:String):Map[TopicAndPartition, Long] = {val offsets = mutable.Map[TopicAndPartition, Long]()val connection = HBaseConnectionPool.getConnectionval tableName = TableName.valueOf("hadoop-topic-offset")val cf = "cf".getBytes()for(topic <- topics) {val rk = s"${topic}-${group}".getBytes()val partition2Offsets = HBaseConnectionPool.getColValue(connection, tableName, rk, cf)partition2Offsets.foreach{case (partition, offset) => {offsets.put(TopicAndPartition(topic, partition), offset)}}}HBaseConnectionPool.release(connection)offsets.toMap} }
查看數(shù)據(jù)結果：

4.3. SparkStreaming和kafka整合需要注意地方

抽取kafka基于zk操作的工具類

package com.desheng.bigdata.streaming.utilimport com.desheng.bigdata.streaming.p2._05KafkaDirectWithZKOps.client import kafka.common.TopicAndPartition import kafka.message.MessageAndMetadata import kafka.serializer.StringDecoder import org.apache.curator.framework.CuratorFramework import org.apache.spark.streaming.StreamingContext import org.apache.spark.streaming.dstream.InputDStream import org.apache.spark.streaming.kafka.{KafkaUtils, OffsetRange}import scala.collection.{JavaConversions, mutable}object KafkaManager {/*從zk中讀取手動保存offset信息，然后從kafka指定offset位置開始讀取數(shù)據(jù)，當然如果沒有讀取到offset信息，那么從最開始或者從最新的位置開始讀取信息*/def createMsg(ssc: StreamingContext, kafkaParams: Map[String, String], topics: Set[String], client:CuratorFramework): InputDStream[(String, String)] = {//從zk中讀取offsetval fromOffsets: Map[TopicAndPartition, Long] = getFromOffsets(topics, kafkaParams("group.id"), client)var messages: InputDStream[(String, String)] = nullif(fromOffsets.isEmpty) {//沒有讀到offsetmessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)} else {//讀取到了offset，從指定位置開始讀取val messageHandler = (msgHandler:MessageAndMetadata[String, String]) => (msgHandler.key(), msgHandler.message())messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc,kafkaParams, fromOffsets, messageHandler)}messages}/*從zk中讀取offset信息首先，定義offset交互的信息，比如數(shù)據(jù)的存放位置，存放格式官方的操作：/kafka/consumers/${group.id}/offsets/${topic}/${partition} -->data為offset存放位置&數(shù)據(jù)：/kafka/consumers/offsets/${topic}/${group.id}/${partition} -->data為offset其次，用什么和zk中交互zookeeper原生的apicuratorFramework(選擇)*/def getFromOffsets(topics:Set[String], group:String, client:CuratorFramework):Map[TopicAndPartition, Long] = {val offsets = mutable.Map[TopicAndPartition, Long]()for (topic <- topics) {val path = s"${topic}/${group}"//判斷當前路徑是否存在checkExists(path, client)for(partition <- JavaConversions.asScalaBuffer(client.getChildren.forPath(path))) {val fullPath = s"${path}/${partition}"val offset = new String(client.getData.forPath(fullPath)).toLongoffsets.put(TopicAndPartition(topic, partition.toInt), offset)}}offsets.toMap}def storeOffsets(offsetRanges: Array[OffsetRange], group:String, client:CuratorFramework) = {for(offsetRange <- offsetRanges) {val topic = offsetRange.topicval partition = offsetRange.partitionval offset = offsetRange.untilOffsetval path = s"${topic}/${group}/${partition}"checkExists(path, client)client.setData().forPath(path, offset.toString.getBytes)}}def checkExists(path:String, client:CuratorFramework): Unit = {if(client.checkExists().forPath(path) == null) {//路徑不能存在client.create().creatingParentsIfNeeded().forPath(path)}} }

4.3.1. 輸出一致性語義的問題

冪等操作
object KafkaOffsetIdempotent {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setAppName("test").setMaster("local[2]")val processingInterval = 2val brokers = "bigdata01:9092,bigdata02:9092,bigdata03:9092"val topic = "mytopic1"// Create direct kafka stream with brokers and topicsval topicsSet = topic.split(",").toSetval kafkaParams = Map[String, String]("metadata.broker.list" -> brokers,"auto.offset.reset" -> "largest","group.id" -> "myspark")/*1. 創(chuàng)建測試的mysql數(shù)據(jù)庫create database test;2. 建表create table myorders(name varchar(20), orderid varchar(100) primary key);3. 新建topic： mytopic1kafka-topics.sh --zookeeper bigdata01:2181/kafka --create --topic mytopic1 --partitions 3 --replication-factor 14. 往mytopic1發(fā)送數(shù)據(jù)，數(shù)據(jù)格式為 "字符,數(shù)字" 比如 abc,3*/val ssc = new StreamingContext(sparkConf, Seconds(processingInterval))val messages = KafkaManager.createMsg(ssc, kafkaParams, topicsSet, client)val jdbcUrl = "jdbc:mysql://localhost:3306/test"val jdbcUser = "root"val jdbcPassword = "sorry"messages.foreachRDD(rdd=>{if(!rdd.isEmpty()) {val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRangesrdd.map(x=>x._2).foreachPartition(partition =>{val pOffsetRange = offsetRanges(TaskContext.get.partitionId)val dbConn = DriverManager.getConnection(jdbcUrl, jdbcUser, jdbcPassword)partition.foreach(msg=>{val name = msg.split(",")(0)val orderid = msg.split(",")(1)val sql = s"insert into myorders(name, orderid) values ('$name', '$orderid') ON DUPLICATE KEY UPDATE name='${name}'"val pstmt = dbConn.prepareStatement(sql)pstmt.execute()})dbConn.close()})KafkaManager.storeOffsets(rdd.asInstanceOf[HasOffsetRanges].offsetRanges, kafkaParams("group.id"), client)}})ssc.start()ssc.awaitTermination()}val client = {val client = CuratorFrameworkFactory.builder().connectString("bigdata01:2181,bigdata02:2181,bigdata03:2181").retryPolicy(new ExponentialBackoffRetry(1000, 3)).namespace("kafka/consumers/offsets").build()client.start()client} }
原子性操作

基于scala-jdbc的方式操作數(shù)據(jù)庫
package com.desheng.bigdata.streaming.p2.exactlyimport kafka.common.TopicAndPartition import kafka.message.MessageAndMetadata import kafka.serializer.StringDecoder import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils} import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.{SparkConf, TaskContext} import scalikejdbc.{ConnectionPool, DB} import scalikejdbc._/*** 事務{* 保存數(shù)據(jù)* 保存offset* }*** 1. 創(chuàng)建測試的mysql數(shù)據(jù)庫create database test;2. 新建topic： mytopic1kafka-topics.sh --zookeeper bigdata01:2181/kafka --create --topic mytopic1 --partitions 3 --replication-factor 13. 建表create table mytopic(topic varchar(200), partid int, offset bigint);create table mydata(name varchar(200), id int);初始化表：insert into mytopic(topic, partid, offset) values('mytopic1',0,0);insert into mytopic(topic, partid, offset) values('mytopic1',1,0);insert into mytopic(topic, partid, offset) values('mytopic1',2,0);4. 往mytopic1發(fā)送數(shù)據(jù)，數(shù)據(jù)格式為 "字符,數(shù)字" 比如 abc,35. 在pom文件加入依賴<dependency><groupId>org.scalikejdbc</groupId><artifactId>scalikejdbc_2.11</artifactId><version>3.2.0</version></dependency>*/ object KafkaOffsetTransanction {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setAppName("test").setMaster("local[2]")val processingInterval = 2val brokers = "bigdata01:9092,bigdata02:9092,bigdata03:9092"val topic = "mytopic1"// Create direct kafka stream with brokers and topicsval topicsSet = topic.split(",").toSetval kafkaParams = Map[String, String]("metadata.broker.list" -> brokers, "auto.offset.reset" -> "smallest")val ssc = new StreamingContext(sparkConf, Seconds(processingInterval))val groupName = "myspark"//val messages = createMyDirectKafkaStream(ssc, kafkaParams, topicsSet, groupName)val driver = "com.mysql.jdbc.Driver"val jdbcUrl = "jdbc:mysql://localhost:3306/test"val jdbcUser = "root"val jdbcPassword = "sorry"// 設置jdbcClass.forName(driver)// 設置連接池ConnectionPool.singleton(jdbcUrl, jdbcUser, jdbcPassword)val fromOffsets = DB.readOnly { implicit session => sql"select topic, partid, offset from mytopic".map { r =>TopicAndPartition(r.string(1), r.int(2)) -> r.long(3)}.list.apply().toMap}val messageHandler = (mmd : MessageAndMetadata[String, String]) => (mmd.topic, mmd.message())val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc, kafkaParams, fromOffsets, messageHandler)messages.foreachRDD(rdd=> {if(!rdd.isEmpty()) {rdd.foreachPartition(partiton=>{val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRangesval pOffsetRange = offsetRanges(TaskContext.get.partitionId)// localTx 開啟事務操作DB.localTx { implicit session =>partiton.foreach(msg=>{// 或者使用scalike的batch 插入val name = msg._2.split(",")(0)val id =msg._2.split(",")(1)val dataResult = sql"""insert into mydata(name,id) values (${name},${id})""".execute().apply()}) // val i = 1 / 0val offsetResult =sql"""update mytopic set offset = ${pOffsetRange.untilOffset} where topic =${pOffsetRange.topic} and partid = ${pOffsetRange.partition}""".update.apply()}})}})ssc.start()ssc.awaitTermination()} }

4.3.2. 常見問題

限流

sparkstreaming要從kafka拉取數(shù)據(jù)，并進行處理；下一次再循環(huán)，如果批次的間隔時間為2s，但是數(shù)據(jù)的處理時間為3s，所以會有越來越多的沒有被處理的數(shù)據(jù)進行累積，最后會拖垮程序，這不是我們所期望的。

解決思路，只能限制流量。非常簡單，通過一個參數(shù)搞定：spark.streaming.kafka.maxRatePerPartition

spark.streaming.kafka.maxRatePerPartition: spark程序每秒中從每個partition分區(qū)讀取的最大的數(shù)據(jù)條數(shù)。比如batchInterval為2s，topic的分區(qū)為3，該參數(shù)的值為100，請問，每個批次最多可以讀取多少條數(shù)據(jù)？2×3×100=600條。
object _07SparkStreamingWithKafkaLimitRateOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("SparkStreamingWithKafkaDirect").setMaster("local").set("spark.streaming.kafka.maxRatePerPartition", "100")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)val kafkaParams = Map[String, String]("bootstrap.servers" -> "bigdata01:9092,bigdata02:9092,bigdata03:9092","group.id" -> "g_1903_2","auto.offset.reset" -> "smallest")val topics = "spark".split(",").toSetval messages: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)messages.foreachRDD((rdd, bTime) => {if(!rdd.isEmpty()) {println("-------------------------------------------")println(s"Time: $bTime")println("########rdd's count: " + rdd.count())println("-------------------------------------------")}})ssc.start()ssc.awaitTermination()} }

偏移量過期（作業(yè)）

4.3.3. 在線ETL應用案例

? 實時的做數(shù)據(jù)的ETL。

需求

數(shù)據(jù)格式：
<<<!>>>3111<<<!>>> --->記錄id <<<!>>>238<<<!>>> --->userid <<<!>>>20181111132902<<<!>>> --->用戶操作時間 <<<!>>>58.223.1.112<<<!>>> --->用戶ip地址 <<<!>>>202.102.92.18<<<!>>> --->用戶訪問ip地址 <<<!>>>59947<<<!>>> --->用戶端口 <<<!>>>80<<<!>>> --->服務端口 <<<!>>>www.sumecjob.com<<<!>>> -->服務域名 <<<!>>><<<!>>> <<<!>>><<<!>>> <<<!>>><<<!>>> <<<!>>><<<!>>> <<<!>>><<<!>>> <<<!>>>http://www.sumecjob.com/Social.aspx<<<!>>> -->用戶訪問的具體url <<<!>>>2556928066<<<!>>>
從kafka指定的topic中讀取如上數(shù)據(jù)，進行清洗，剔除上述無用字段，保留有用信息，包括userid 用戶操作時間(timestamp) 用戶ip地址:端口服務地址：服務端口 url,最后將清洗結果送回kafka指定的topic中，完成在線etl。
topic構建
--原始數(shù)據(jù)的topic kafka-topics.sh --create \ --topic orginal \ --zookeeper bigdata01:2181/kafka \ --partitions 3 --replication-factor 1--標準數(shù)據(jù)的topic kafka-topics.sh --create \ --topic standard \ --zookeeper bigdata01:2181/kafka \ --partitions 3 --replication-factor 1
編碼
object _08SparkStreamingWithKafkaETLOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("KafkaDirectWithRedis").setMaster("local[*]")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)val kafkaParams = Map[String, String]("bootstrap.servers" -> "bigdata01:9092,bigdata02:9092,bigdata03:9092","group.id" -> "g_1903_2","auto.offset.reset" -> "smallest")val topics = "source".split(",").toSetval target = "target"val messages: InputDStream[(String, String)] = JedisManager.createMsg(ssc, kafkaParams, topics)messages.foreachRDD((rdd, bTime) => {if(!rdd.isEmpty()) {println("-------------------------------------------")println(s"Time: $bTime")processETL(rdd, target)//處理過程JedisManager.storeOffsets(rdd.asInstanceOf[HasOffsetRanges].offsetRanges,kafkaParams("group.id"))println("-------------------------------------------")}})ssc.start()ssc.awaitTermination()}//在線etl處理過程def processETL(rdd: RDD[(String, String)], targetTopic: String): Unit = {val prcessedRDD = rdd.map{case (key, value) => {val fields = value.split(",")if(fields == null || fields.length != 15) {"" //異常數(shù)據(jù)} else {//正常數(shù)據(jù)val userid = fields(1).replaceAll("<<<!>>>", "")//2018 11 11 13 29 02-->yyyyMMddHHmmss-->yyyy-MM-dd HH:mm:ssval time = fields(2).replaceAll("<<<!>>>", "")val timestamp = DateUtil.time2Stamp(time)val clientIp = fields(3).replaceAll("<<<!>>>", "")val serverIp = fields(4).replaceAll("<<<!>>>", "")val clientPort = fields(5).replaceAll("<<<!>>>", "")val serverPort = fields(6).replaceAll("<<<!>>>", "")val url = fields(13).replaceAll("<<<!>>>", "")s"${userid}|${timestamp}|${clientIp}:${clientPort}|${serverIp}:${serverPort}|${url}"}}}.filter(_ != "")//將清洗好的數(shù)據(jù)送回kafka中 // produceInfo1(cleanedRDD, destTopic)produceInfo2(cleanedRDD, destTopic)}def produceInfo2(cleanedRDD: RDD[String], destTopic:String): Unit = {cleanedRDD.foreachPartition(partition => {if(!partition.isEmpty) {val conf = new Properties()conf.put("bootstrap.servers", "bigdata01:9092,bigdata02:9092,bigdata03:9093")conf.put("acks", "1")conf.put("retries", "0")conf.put("batch.size", "16384")conf.put("linger.ms", "5000")conf.put("buffer.memory", "33554432")conf.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")conf.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")val producer = new KafkaProducer[String, String](conf)partition.foreach{line => {val record = new ProducerRecord[String, String](destTopic, line)producer.send(record)}}producer.close()}})}//這種操作效率太低，會創(chuàng)建大量的producerdef produceInfo1(cleanedRDD: RDD[String], destTopic:String): Unit = {cleanedRDD.foreach(line => {val conf = new Properties()conf.put("bootstrap.servers", "bigdata01:9092,bigdata02:9092,bigdata03:9093")conf.put("acks", "1")conf.put("retries", "0")conf.put("batch.size", "16384")conf.put("linger.ms", "5000")conf.put("buffer.memory", "33554432")conf.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")conf.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")val producer = new KafkaProducer[String, String](conf)val record = new ProducerRecord[String, String](destTopic, line)producer.send(record)producer.close()})} }
redis工具類
package com.desheng.bigdata.streaming.utilimport com.desheng.bigdata.common.db.JedisUtil import kafka.common.TopicAndPartition import kafka.message.MessageAndMetadata import kafka.serializer.StringDecoder import org.apache.spark.streaming.StreamingContext import org.apache.spark.streaming.dstream.InputDStream import org.apache.spark.streaming.kafka.{KafkaUtils, OffsetRange}import scala.collection.mutable import scala.collection.JavaConversions._ object JedisManager {def createMsg(ssc: StreamingContext, kafkaParams: Map[String, String], topics: Set[String]): InputDStream[(String, String)] = {//從zk中讀取offsetval fromOffsets: Map[TopicAndPartition, Long] = getFromOffsets(topics)var messages: InputDStream[(String, String)] = nullif(fromOffsets.isEmpty) {//沒有讀到offsetmessages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)} else {//讀取到了offset，從指定位置開始讀取val messageHandler = (msgHandler:MessageAndMetadata[String, String]) => (msgHandler.key(), msgHandler.message())messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](ssc,kafkaParams, fromOffsets, messageHandler)}messages}/*從redis中讀取offset信息K-V首先，定義offset交互的信息，比如數(shù)據(jù)的存放位置，存放格式topic partition offset group五種數(shù)據(jù)結構：stringtopic|group|partition offsetlistabchashkey field-value ...topic group|partition offsetgroup topic|partition offsetsetzset其次，用什么和redis中交互*/def getFromOffsets(topics:Set[String]):Map[TopicAndPartition, Long] = {val offsets = mutable.Map[TopicAndPartition, Long]()val jedis = JedisUtil.getJedisfor (topic <- topics) {val gpo = jedis.hgetAll(topic)for((gp, offset) <- gpo) {val partition = gp.substring(gp.indexOf("|") + 1).toIntoffsets.put(TopicAndPartition(topic, partition), offset.toLong)}}JedisUtil.release(jedis)offsets.toMap}def storeOffsets(offsetRanges: Array[OffsetRange], group:String) = {val jedis = JedisUtil.getJedisfor(offsetRange <- offsetRanges) {val topic = offsetRange.topicval partition = offsetRange.partitionval offset = offsetRange.untilOffsetval gp = s"${group}|${partition}"jedis.hset(topic, gp, offset.toString)}JedisUtil.release(jedis)}

}
```

日期工具類```scala public class DateUtil {private static DateFormat df = new SimpleDateFormat("yyyyMMddHHmmss");//2018 11 11 13 29 02-->yyyyMMddHHmmss-->timestamppublic static long time2Stamp(String srcDate) {try {return df.parse(srcDate).getTime();} catch (ParseException e) {e.printStackTrace();}return System.currentTimeMillis();} } ```在代碼執(zhí)行過程中，可能會出現(xiàn)如下異常：![1565342525589](assets/1565342525589.png)原因是因為使用了線程非安全的SimpleDateFormat造成的，如何解決！以如下案例來解決：```java /*演示SimpleDateFormat多線程安全問題執(zhí)行一下代碼，可能會出現(xiàn)NumberFormatException: For input string: "E.178E1" 主要原因是因為由于SimpleDateFormat為線程非安全的類所造成的*/

public class SampleDateFormatTest {
public static void main(String[] args) {
List list = Arrays.asList(
“2019-08-09 17:12:47”,
“2019-08-08 17:12:47”,
“2019-08-07 17:12:47”,
“2019-08-06 17:12:47”,
“2019-08-05 17:12:47”
);
//轉化為實踐戳，但是以多線程方式操作
SimpleDateFormat df = new SimpleDateFormat(“yyyy-MM-dd HH:mm:ss”);

for (String time : list) {new Thread(() -> {try {System.out.println(df.parse(time));} catch (ParseException e) {e.printStackTrace();}}).start();}} } ```有三種解決方案：1. 使用局部變量創(chuàng)建SimpleDateFormat2. 使用ThreadLocal3. 使用線程安全的DateFormat---FastDateFormat使用apache提供的線程安全的類來處理：org.apache.commons.lang3.time.FastDateFormat思考題：解決sparkstreaming寫入kafka效率問題！

5. SparkStreaming常見transformation算子

5.1. 常見的算子操作

? 對cogroup做一簡單的說明：cogroup就是groupByKey的另外一種變體，groupByKey是操作一個K-V鍵值對，而cogroup一次操作兩個，有點像join，不同之處在于返回值結果：

val ds1:DStream[(K, V)] val ds2:DStream[(K, w)] val cg:DStream[(K, (Iterable[V], Iterable[W]))] = ds1.cogroup(ds1)

5.2. transform

5.2.1. 概述

? transform是一個transformation算子，轉換算子。

? 人話：DStream上述提供的所有的transformation操作，都是DStream-2-DStream操作，沒有一個DStream和RDD的直接操作，而DStream本質上是一系列RDD，所以RDD-2-RDD操作是顯然被需要的，所以此時官方api中提供了一個為了達成此操作的算子——transform操作。

? 其最最最經(jīng)典的實現(xiàn)就是DStream和rdd的join操作，還有dstream重分區(qū)(分區(qū)減少，coalsce)。

? 也就是說transform主要就是用來自定義官方api沒有提供的一些操作。

5.2.2. 需求簡介——動態(tài)黑名單過濾

? 廣告計費系統(tǒng)，是電商必不可少的一個功能點。為了防止惡意的廣告點擊(假設商戶A和B同時在某電商做了廣告，A和B為競爭對手，那么如果A使用點擊機器人進行對B的廣告的惡意點擊，那么B的廣告費用將很快被用完)，必須對廣告點擊進行黑名單過濾。黑名單的過濾可以是ID，可以是IP等等，黑名單就是過濾的條件，利用SparkStreaming的流處理特性，可實現(xiàn)實時黑名單的過濾實現(xiàn)。可以使用leftouter join 對目標數(shù)據(jù)和黑名單數(shù)據(jù)進行關聯(lián)，將命中黑名單的數(shù)據(jù)過濾掉。

5.2.3. 代碼實現(xiàn)

/*** 在線黑名單過濾** 類名起名規(guī)范* 首字母大寫，多單詞，采用駝峰* 一律名詞，不能動詞* 并且單數(shù)不能復數(shù)* 方法名起名規(guī)范* 首字母小寫，多單詞，采用駝峰* 一般采用動賓短語（動詞+名詞）* 盡量少用一些漢語拼音，中文** 需求：* 從用戶請求的nginx日志中過濾出黑名單的數(shù)據(jù)，保留白名單數(shù)據(jù)進行后續(xù)業(yè)務統(tǒng)計。* data structure* 27.19.74.143##2016-05-30 17:38:20##GET /static/image/common/faq.gif HTTP/1.1##200##1127 110.52.250.126##2016-05-30 17:38:20##GET /data/cache/style_1_widthauto.css?y7a HTTP/1.1##200##1292*/ object _01OnlineBlacklistFilterOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("OnlineBlacklistFilter").setMaster("local[*]")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)//黑名單RDDval blacklistRDD:RDD[(String, Boolean)] = ssc.sparkContext.parallelize(List(("27.19.74.143", true),("110.52.250.126", true)))//接入外部的數(shù)據(jù)流val lines:DStream[String] = ssc.socketTextStream("bigdata01", 9999)//黑名單過濾 // 110.52.250.126##2016-05-30 17:38:20##GET /data/cache/style_1_widthauto.css?y7a HTTP/1.1##200##1292val ip2OtherDStream:DStream[(String, String)] = lines.map(line => {val index = line.indexOf("##")val ip = line.substring(0, index)val other = line.substring(index + 2)(ip, other)})val filteredDStream:DStream[(String, String)] = ip2OtherDStream.transform(rdd => {val join = rdd.leftOuterJoin(blacklistRDD)join.filter{case (ip, (left, right)) => {!right.isDefined}}.map{case (ip, (left, right)) => {(ip, left)}}})filteredDStream.print()//重分區(qū) // filteredDStream.transform(_.coalesce(8))ssc.start()ssc.awaitTermination()} }

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-Uh6hF29B-1587103398584)(assets/1565575184937.png)]

5.3. updateStateByKey

5.3.1. 概述

? updateStateByKey(func) 根據(jù)于key的前置狀態(tài)和key的新值，對key進行更新，返回一個新狀態(tài)的Dstream。

? 人話：統(tǒng)計截止到目前為止key的狀態(tài)。

? 通過分析，我們需要清楚：在這個操作中需要兩個數(shù)據(jù)，一個是key的前置狀態(tài)，一個是key的新增(當前批次的數(shù)據(jù))；還有歷史數(shù)據(jù)(前置狀態(tài))得需要存儲在磁盤，不應該保存在內存中。

? 同時key的前置狀態(tài)可能有可能沒有。

5.3.2. 案例實現(xiàn)——wordcount

/*** 統(tǒng)計，截止到目前為止出現(xiàn)的每一個key的次數(shù)*/ object _02WordCountUpdateStateByKeyOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("WordCountUpdateStateByKey").setMaster("local[*]")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)ssc.checkpoint("file:/E:/data/out/1903/chk")val lines:DStream[String] = ssc.socketTextStream("bigdata01", 9999)val pairs:DStream[(String, Int)] = lines.flatMap(_.split("\\s+")).map((_, 1))val usb:DStream[(String, Int)] = pairs.updateStateByKey(updateFunc)usb.print()ssc.start()ssc.awaitTermination()}/*狀態(tài)更新函數(shù)根據(jù)key的前置狀態(tài)和key的最新值，聚合得到截止到目前為止key的狀態(tài)seq:為當前key的狀態(tài)option為key對應的歷史值*/def updateFunc(seq: Seq[Int], option: Option[Int]): Option[Int] = {println("option:" + option + "> seq: " + seq.mkString("[", ",", "]")) // var sum = 0 // for(i <- seq) sum += i // if(option.isDefined) { // sum += option.get // } // Option(sum)Option(seq.sum + option.getOrElse(0))} }

[外鏈圖片轉存失敗,源站可能有防盜鏈機制,建議將圖片保存下來直接上傳(img-cd5fOxU8-1587103398587)(assets/1565577345261.png)]

5.4. window

5.4.1. 概述

? window操作就是窗口函數(shù)。Spark Streaming提供了滑動窗口操作的支持，從而讓我們可以對一個滑動窗口內的數(shù)據(jù)執(zhí)行計算操作。每次掉落在窗口內的RDD的數(shù)據(jù)，會被聚合起來執(zhí)行計算操作，然后生成的RDD，會作為window DStream的一個RDD。比如下圖中，就是對每三秒鐘的數(shù)據(jù)執(zhí)行一次滑動窗口計算，這3秒內的3個RDD會被聚合起來進行處理，然后過了兩秒鐘，又會對最近三秒內的數(shù)據(jù)執(zhí)行滑動窗口計算。所以每個滑動窗口操作，都必須指定兩個參數(shù)，窗口長度以及滑動間隔，而且這兩個參數(shù)值都必須是batch間隔的整數(shù)倍。

紅色的矩形就是一個窗口，窗口hold的是一段時間內的數(shù)據(jù)流。

這里面每一個time都是時間單元，在官方的例子中，每隔window size是3 time unit, 而且每隔2個單位時間，窗口會slide一次。
所以基于窗口的操作，需要指定2個參數(shù)：
window length - The duration of the window (3 in the figure)
slide interval - The interval at which the window-based operation is performed (2 in the figure).

窗口大小，個人感覺是一段時間內數(shù)據(jù)的容器。

滑動間隔，就是我們可以理解的cron表達式吧。

5.4.2. 案例說明

/*** 統(tǒng)計，截止到目前為止出現(xiàn)的每一個key的次數(shù)* window窗口操作，每個多長M時間,通過過往N長時間內產生的數(shù)據(jù)* M就是滑動長度sliding interval* N就是窗口長度window length*/ object _03WordCountWindowsOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("WordCountUpdateStateByKey").setMaster("local[*]")val batchInterval = 2val duration = Seconds(batchInterval)val ssc = new StreamingContext(conf, duration)val lines:DStream[String] = ssc.socketTextStream("bigdata01", 9999)val pairs:DStream[(String, Int)] = lines.flatMap(_.split("\\s+")).map((_, 1))val ret:DStream[(String, Int)] = pairs.reduceByKeyAndWindow(_+_,windowDuration = Seconds(batchInterval * 3),slideDuration = Seconds(batchInterval * 2))ret.print()ssc.start()ssc.awaitTermination()}/*狀態(tài)更新函數(shù)根據(jù)key的前置狀態(tài)和key的最新值，聚合得到截止到目前為止key的狀態(tài)seq:為當前key的狀態(tài)option為key對應的歷史值*/def updateFunc(seq: Seq[Int], option: Option[Int]): Option[Int] = {println("option:" + option + "> seq: " + seq.mkString("[", ",", "]")) // var sum = 0 // for(i <- seq) sum += i // if(option.isDefined) { // sum += option.get // } // Option(sum)Option(seq.sum + option.getOrElse(0))} }

6. SparkSQL和SparkStreaming的整合案例

6.1. 需求

? Spark最強大的地方在于，可以與Spark Core、Spark SQL整合使用，之前已經(jīng)通過transform、foreachRDD等算子看到，如何將DStream中的RDD使用Spark Core執(zhí)行批處理操作。現(xiàn)在就來看看，如何將DStream中的RDD與Spark SQL結合起來使用。
案例：top3的商品排序：最新的top3

? 這里就是基于updatestateByKey，統(tǒng)計截止到目前為止的不同品類下的商品銷量top3

6.2. 代碼實現(xiàn)

/*** SparkStreaming整合SparkSQL的案例之，熱門品類top3排行*/ object _04StreamingIntegerationSQLOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("StreamingIntegerationSQL").setMaster("local[*]")val batchInterval = 2val duration = Seconds(batchInterval)val spark = SparkSession.builder().config(conf).getOrCreate()val ssc = new StreamingContext(spark.sparkContext, duration)ssc.checkpoint("file:/E:/data/out/1903/chk-1")val lines:DStream[String] = ssc.socketTextStream("bigdata01", 9999)//001 mi moblieval pairs:DStream[(String, Int)] = lines.map(line => {val fields = line.split("\\s+")if(fields == null || fields.length != 3) {("", -1)} else {val brand = fields(1)val category = fields(2)(s"${category}_${brand}", 1)}}).filter(t => t._2 != -1)val usb:DStream[(String, Int)] = pairs.updateStateByKey(updateFunc)usb.foreachRDD((rdd, bTime) => {if(!rdd.isEmpty()) {//category_brand countimport spark.implicits._val df = rdd.map{case (cb, count) => {val category = cb.substring(0, cb.indexOf("_"))val brand = cb.substring(cb.indexOf("_") + 1)(category, brand, count)}}.toDF("category", "brand", "sales")df.createOrReplaceTempView("tmp_category_brand_sales")val sql ="""|select| t.category,| t.brand,| t.sales| t.rank|from (| select| category,| brand,| sales,| row_number() over(partition by category order by sales desc) rank| from tmp_category_brand_sales|) t|where t.rank < 4""".stripMarginspark.sql(sql).show()}})ssc.start()ssc.awaitTermination()}def updateFunc(seq: Seq[Int], option: Option[Int]): Option[Int] = {Option(seq.sum + option.getOrElse(0))} }

7. SparkStreaming高可用及其優(yōu)化建議

7.1. SparkStreaming緩存操作

? SparkStreaming的緩存，說白了就是DStream的緩存，DStream的緩存就只有一個方面，DStream對應的RDD的緩存，RDD如何緩存？rdd.persist()，所以DStream的緩存說白了就是RDD的緩存，使用persist()指定，及其需要指定持久化策略，大多算子默認情況下，持久化策略為MEMORY_AND_DISK_SER_2。

7.2. SparkStreaming的checkpoint機制

7.2.1 概述

1、每一個Spark Streaming應用，正常來說，都是要7*24小時運轉的，這就是實時計算程序的特點。因為要持續(xù)不斷的對數(shù)據(jù)進行計算。因此，對實時計算應用的要求，應該是必須要能夠對與應用程序邏輯無關的失敗，進行容錯。
2、如果要實現(xiàn)這個目標，Spark Streaming程序就必須將足夠的信息checkpoint到容錯的存儲系統(tǒng)上，從而讓它能夠從失敗中進行恢復。有兩種數(shù)據(jù)需要被進行checkpoint：
1）元數(shù)據(jù)checkpoint——將定義了流式計算邏輯的信息，保存到容錯的存儲系統(tǒng)上，比如HDFS。當運行Spark Streaming應用程序的Driver進程所在節(jié)點失敗時，該信息可以用于進行恢復。元數(shù)據(jù)信息包括了：
配置信息——創(chuàng)建Spark Streaming應用程序的配置信息，比如SparkConf中的信息。
DStream的操作信息——定義了Spark Stream應用程序的計算邏輯的DStream操作信息。
未處理的batch信息——那些job正在排隊，還沒處理的batch信息。

2)、數(shù)據(jù)checkpoint——將實時計算過程中產生的RDD的數(shù)據(jù)保存到可靠的存儲系統(tǒng)中。
對于一些將多個batch的數(shù)據(jù)進行聚合的，有狀態(tài)的transformation操作，這是非常有用的。在這種transformation操作中，生成的RDD是依賴于之前的batch的RDD的，這會導致隨著時間的推移，RDD的依賴鏈條變得越來越長。
要避免由于依賴鏈條越來越長，導致的一起變得越來越長的失敗恢復時間，有狀態(tài)的transformation操作執(zhí)行過程中間產生的RDD，會定期地被checkpoint到可靠的存儲系統(tǒng)上，比如HDFS。從而削減RDD的依賴鏈條，進而縮短失敗恢復時，RDD的恢復時間。
總結，元數(shù)據(jù)checkpoint主要是為了從driver失敗中進行恢復；而RDD checkpoint主要是為了，使用到有狀態(tài)的transformation操作時，能夠在其生產出的數(shù)據(jù)丟失時，進行快速的失敗恢復。

7.2.2. 啟動checkpoint

啟動方式一

? 1、使用了有狀態(tài)的transformation操作——比如updateStateByKey，或者reduceByKeyAndWindow操作，被使用了，那么checkpoint目錄要求是必須提供的，也就是必須開啟checkpoint機制，從而進行周期性的RDD checkpoint。
? 2、要保證可以從Driver失敗中進行恢復——元數(shù)據(jù)checkpoint需要啟用，來進行這種情況的恢復。
? 3、要注意的是，并不是說，所有的Spark Streaming應用程序，都要啟用checkpoint機制，如果即不強制要求從Driver失敗中自動進行恢復，又沒使用有狀態(tài)的transformation操作，那么就不需要啟用checkpoint。事實上，這么做反而是有助于提升性能的。

啟動方式二

? 1、對于有狀態(tài)的transformation操作，啟用checkpoint機制，定期將其生產的RDD數(shù)據(jù)checkpoint，是比較簡單的。
可以通過配置一個容錯的、可靠的文件系統(tǒng)（比如HDFS）的目錄，來啟用checkpoint機制，checkpoint數(shù)據(jù)就會寫入該目錄。使用StreamingContext的checkpoint()方法即可。然后，你就可以放心使用有狀態(tài)的transformation操作了。
? 2、如果為了要從Driver失敗中進行恢復，那么啟用checkpoint機制，是比較復雜的。需要改寫Spark Streaming應用程序。
? 當應用程序第一次啟動的時候，需要創(chuàng)建一個新的StreamingContext，并且調用其start()方法，進行啟動。當Driver從失敗中恢復過來時，需要從checkpoint目錄中記錄的元數(shù)據(jù)中，恢復出來一個StreamingContext。

? 這里針對第二點(重新修改代碼)做一說明：

def createFuc():StreamingContext = {val ssc = new StreamingContext(conf, batchInterval)ssc.checkpoint(checkpoint)//業(yè)務邏輯.....ssc } val ssc = StreamingContext.getOrCreate(checkpoint, createFunc)

eg:

/*offset的checkpoint(檢查點)把需要管理的相關數(shù)據(jù)保存在某一個目錄下面，后續(xù)的時候直接從該目錄中讀取即可，在此處就是保存offset數(shù)據(jù)*/ object _04CheckpointWithKafkaDirectOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)val conf = new SparkConf().setAppName("CheckpointWithKafkaDirectOps").setMaster("local")val duration = Seconds(2)val checkpoint = "file:///E:/data/monitored/chk"def createFunc():StreamingContext = {val ssc = new StreamingContext(conf, duration)ssc.checkpoint(checkpoint)val kafkaParams = Map[String, String]("bootstrap.servers" -> "bigdata01:9092,bigdata02:9092,bigdata03:9092","group.id" -> "g_1903_2","auto.offset.reset" -> "smallest")val topics = "spark".split(",").toSetval messages: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)messages.foreachRDD((rdd, bTime) => {if(!rdd.isEmpty()) {println("num: " + rdd.getNumPartitions)val offsetRDD = rdd.asInstanceOf[HasOffsetRanges]val offsetRanges = offsetRDD.offsetRangesfor(offsetRange <- offsetRanges) {val topic = offsetRange.topicval partition = offsetRange.partitionval fromOffset = offsetRange.fromOffsetval untilOffset = offsetRange.untilOffsetprintln(s"topic:${topic}\tpartition:${partition}\tstart:${fromOffset}\tend:${untilOffset}")}rdd.count()}})ssc}//創(chuàng)建或者恢復出來一個StreamingContextval ssc = StreamingContext.getOrCreate(checkpoint, createFunc)ssc.start()ssc.awaitTermination()} }

? 當程序對應的driver失敗進行恢復的時候，上述的修改，只是完成了第一步，還有第二步，第三步要走。

第二步，修改spark-submit腳本中的參數(shù)：–deploy-mode cluster

第三步，修改spark-submit腳本中的參數(shù)：–supervise

7.3. DriverHA

7.3.1. DriverHA的原理

由于流計算系統(tǒng)是長期運行、且不斷有數(shù)據(jù)流入，因此其Spark守護進程（Driver）的可靠性至關重要，它決定了Streaming程序能否一直正確地運行下去。Driver實現(xiàn)HA的解決方案就是將元數(shù)據(jù)持久化，以便重啟后的狀態(tài)恢復。如圖一所示，Driver持久化的元數(shù)據(jù)包括： Block元數(shù)據(jù)（圖1中的綠色箭頭）：Receiver從網(wǎng)絡上接收到的數(shù)據(jù)，組裝成Block后產生的Block元數(shù)據(jù)；

Checkpoint數(shù)據(jù)（圖1中的橙色箭頭）：包括配置項、DStream操作、未完成的Batch狀態(tài)、和生成的RDD數(shù)據(jù)等；

恢復計算（圖2中的橙色箭頭）：使用Checkpoint數(shù)據(jù)重啟driver，重新構造上下文并重啟接收器。恢復元數(shù)據(jù)塊（圖2中的綠色箭頭）：恢復Block元數(shù)據(jù)。恢復未完成的作業(yè)（圖2中的紅色箭頭）：使用恢復出來的元數(shù)據(jù)，再次產生RDD和對應的job，然后提交到Spark集群執(zhí)行。通過如上的數(shù)據(jù)備份和恢復機制，Driver實現(xiàn)了故障后重啟、依然能恢復Streaming任務而不丟失數(shù)據(jù)，因此提供了系統(tǒng)級的數(shù)據(jù)高可靠。

7.3.2. DriverHA的配置

#!/bin/shSPARK_HOME=/home/bigdata/app/spark$SPARK_HOME/bin/spark-submit \ --master spark://bigdata01:7077 \ --deploy-mode cluster \ --class com.desheng.bigdata.streaming.p3._05SparkStreamingDriverHAOps \ --executor-memory 600M \ --executor-cores 2 \ --driver-cores 1 \ --supervise \ --total-executor-cores 3 \ hdfs://ns1/jars/spark/1903-bd/sparkstreaming-drverha.jar \ 2 bigdata01 9999 \ hdfs://ns1/checkpoint/spark/driverha

7.3.3. Driver代碼實現(xiàn)

object _05SparkStreamingDriverHAOps {def main(args: Array[String]): Unit = {Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)Logger.getLogger("org.apache.spark").setLevel(Level.WARN)Logger.getLogger("org.spark_project").setLevel(Level.WARN)if(args == null || args.length < 4) {System.err.println("""|Parameter Errors! Usage: <batchInterval> <host> <port> <checkpoint>""".stripMargin)System.exit(-1)}val Array(batchInterval, host, port, checkpoint) = argsval conf = new SparkConf().setAppName("SparkStreamingDriverHA").setMaster("local[*]")def createFunc():StreamingContext = {val ssc = new StreamingContext(conf, Seconds(batchInterval.toLong))ssc.checkpoint(checkpoint)val lines:DStream[String] = ssc.socketTextStream(host, port.toInt)val pairs:DStream[(String, Int)] = lines.flatMap(_.split("\\s+")).map((_, 1))val usb:DStream[(String, Int)] = pairs.updateStateByKey((seq, option) => Option(seq.sum + option.getOrElse(0)))usb.print()ssc}val ssc = StreamingContext.getOrCreate(checkpoint, createFunc)ssc.start()ssc.awaitTermination()} }

7.4. SparkStreaming程序的部署、升級與維護

部署

升級

監(jiān)控

7.5. 調優(yōu)建議

7.5.1.設置合理的CPU

很多情況下Streaming程序需要的內存不是很多，但是需要的CPU要很多。在Streaming程序中，CPU資源的使用可以分為兩大類：

（1）、用于接收數(shù)據(jù)；

（2）、用于處理數(shù)據(jù)。我們需要設置足夠的CPU資源，使得有足夠的CPU資源用于接收和處理數(shù)據(jù)，這樣才能及時高效地處理數(shù)據(jù)。

7.5.2.關于接受數(shù)據(jù)的調優(yōu)說明

1、通過網(wǎng)絡接收數(shù)據(jù)時（比如Kafka、Flume、ZMQ、RocketMQ、RabbitMQ和ActiveMQ等），會將數(shù)據(jù)反序列化，并存儲在Spark的內存中。

2、如果數(shù)據(jù)接收成為系統(tǒng)的瓶頸，那么可以考慮并行化數(shù)據(jù)接收。每一個輸入DStream都會在某個Worker的Executor上啟動一個Receiver，該Receiver接收一個數(shù)據(jù)流。因此可以通過創(chuàng)建多個輸入DStream，并且配置它們接收數(shù)據(jù)源不同的分區(qū)數(shù)據(jù)，達到接收多個數(shù)據(jù)流的效果。

3、舉例說明：一個接收4個Kafka Topic的輸入DStream，可以被拆分為兩個輸入DStream，每個分別接收二個topic的數(shù)據(jù)。這樣就會創(chuàng)建兩個Receiver，從而并行地接收數(shù)據(jù)，進而提升吞吐量。多個DStream可以使用union算子進行聚合，從而形成一個DStream。然后后續(xù)的transformation算子操作都針對該一個聚合后的DStream即可。

4、使用inputStream.repartition()即可。這樣就可以將接收到的batch，分布到指定數(shù)量的機器上，然后再進行進一步的操作。

5、數(shù)據(jù)接收并行度調優(yōu)，除了創(chuàng)建更多輸入DStream和Receiver以外，還可以考慮調節(jié)block interval。通過參數(shù)，spark.streaming.blockInterval，可以設置block interval，默認是200ms。對于大多數(shù)Receiver來說，在將接收到的數(shù)據(jù)保存到Spark的BlockManager之前，都會將數(shù)據(jù)切分為一個一個的block。而每個batch中的block數(shù)量，則決定了該batch對應的RDD的partition的數(shù)量，以及針對該RDD執(zhí)行transformation操作時，創(chuàng)建的task的數(shù)量。每個batch對應的task數(shù)量是大約估計的，即batch interval / block interval。

舉個例子

1)、batch interval為3s，block interval為150ms，會創(chuàng)建20個task。如果你認為每個batch的task數(shù)量太少，即低于每臺機器的cpu core數(shù)量，那么就說明batch的task數(shù)量是不夠的，因為所有的cpu資源無法完全被利用起來。要為batch增加block的數(shù)量，那么就減小block interval

2)、推薦的block interval最小值是50ms，如果低于這個數(shù)值，那么大量task的啟動時間，可能會變成一個性能開銷點。

7.5.2.設置合理的并行度

如果在計算的任何stage中使用的并行task的數(shù)量沒有足夠多，那么集群資源是無法被充分利用的。舉例來說，對于分布式的reduce操作，比如reduceByKey和reduceByKeyAndWindow，默認的并行task的數(shù)量是由spark.default.parallelism參數(shù)決定的。你可以在reduceByKey等操作中，傳入第二個參數(shù)，手動指定該操作的并行度，也可以調節(jié)全局的spark.default.parallelism參數(shù)

該參數(shù)說的是，對于那些shuffle的父RDD的最大的分區(qū)數(shù)據(jù)。對于parallelize或者textFile這些輸入算子，因為沒有父RDD，所以依賴于ClusterManager的配置。如果是local模式，該默認值是local[x]中的x；如果是mesos的細粒度模式，該值為8，其它模式就是Math.max(2, 所有的excutor上的所有的core的總數(shù))。

7.5.3.序列化調優(yōu)說明

數(shù)據(jù)序列化造成的系統(tǒng)開銷可以由序列化格式的優(yōu)化來減小。在流式計算的場景下，有兩種類型的數(shù)據(jù)需要序列化。
1、輸入數(shù)據(jù)：默認情況下，接收到的輸入數(shù)據(jù)，是存儲在Executor的內存中的，使用的持久化級別是StorageLevel.MEMORY_AND_DISK_SER_2。這意味著，數(shù)據(jù)被序列化為字節(jié)從而減小GC開銷，并且會復制以進行executor失敗的容錯。因此，數(shù)據(jù)首先會存儲在內存中，然后在內存不足時會溢寫到磁盤上，從而為流式計算來保存所有需要的數(shù)據(jù)。這里的序列化有明顯的性能開銷——Receiver必須反序列化從網(wǎng)絡接收到的數(shù)據(jù)，然后再使用Spark的序列化格式序列化數(shù)據(jù)。

2、流式計算操作生成的持久化RDD：流式計算操作生成的持久化RDD，可能會持久化到內存中。例如，窗口操作默認就會將數(shù)據(jù)持久化在內存中，因為這些數(shù)據(jù)后面可能會在多個窗口中被使用，并被處理多次。然而，不像Spark Core的默認持久化級別，StorageLevel.MEMORY_ONLY，流式計算操作生成的RDD的默認持久化級別是StorageLevel.MEMORY_ONLY_SER ，默認就會減小GC開銷。

在上述的場景中，使用Kryo序列化類庫可以減小CPU和內存的性能開銷。使用Kryo時，一定要考慮注冊自定義的類，并且禁用對應引用的tracking（spark.kryo.referenceTracking）。

7.5.4. batchInterval

如果想讓一個運行在集群上的Spark Streaming應用程序可以穩(wěn)定，它就必須盡可能快地處理接收到的數(shù)據(jù)。換句話說，batch應該在生成之后，就盡可能快地處理掉。對于一個應用來說，這個是不是一個問題，可以通過觀察Spark UI上的batch處理時間來定。batch處理時間必須小于batch interval時間。

在構建StreamingContext的時候，需要我們傳進一個參數(shù)，用于設置Spark Streaming批處理的時間間隔。Spark會每隔batchDuration時間去提交一次Job，如果你的Job處理的時間超過了batchDuration的設置，那么會導致后面的作業(yè)無法按時提交，隨著時間的推移，越來越多的作業(yè)被拖延，最后導致整個Streaming作業(yè)被阻塞，這就間接地導致無法實時處理數(shù)據(jù)，這肯定不是我們想要的。

另外，雖然batchDuration的單位可以達到毫秒級別的，但是經(jīng)驗告訴我們，如果這個值過小將會導致因頻繁提交作業(yè)從而給整個Streaming帶來負擔，所以請盡量不要將這個值設置為小于500ms。在很多情況下，設置為500ms性能就很不錯了。

那么，如何設置一個好的值呢？我們可以先將這個值位置為比較大的值（比如10S），如果我們發(fā)現(xiàn)作業(yè)很快被提交完成，我們可以進一步減小這個值，知道Streaming作業(yè)剛好能夠及時處理完上一個批處理的數(shù)據(jù)，那么這個值就是我們要的最優(yōu)值。

7.5.5. 內存調優(yōu)

內存調優(yōu)的另外一個方面是垃圾回收。對于流式應用來說，如果要獲得低延遲，肯定不想要有因為JVM垃圾回收導致的長時間延遲。有很多參數(shù)可以幫助降低內存使用和GC開銷：

1、DStream的持久化：正如在“數(shù)據(jù)序列化調優(yōu)”一節(jié)中提到的，輸入數(shù)據(jù)和某些操作生產的中間RDD，默認持久化時都會序列化為字節(jié)。與非序列化的方式相比，這會降低內存和GC開銷。使用Kryo序列化機制可以進一步減少內存使用和GC開銷。進一步降低內存使用率，可以對數(shù)據(jù)進行壓縮，由spark.rdd.compress參數(shù)控制（默認false）。

2、清理舊數(shù)據(jù)：默認情況下，所有輸入數(shù)據(jù)和通過DStream transformation操作生成的持久化RDD，會自動被清理。Spark Streaming會決定何時清理這些數(shù)據(jù)，取決于transformation操作類型。例如，你在使用窗口長度為10分鐘內的window操作，Spark會保持10分鐘以內的數(shù)據(jù)，時間過了以后就會清理舊數(shù)據(jù)。但是在某些特殊場景下，比如Spark SQL和Spark Streaming整合使用時，在異步開啟的線程中，使用Spark SQL針對batch RDD進行執(zhí)行查詢。那么就需要讓Spark保存更長時間的數(shù)據(jù)，直到Spark SQL查詢結束。可以使用streamingContext.remember()方法來實現(xiàn)。

? 3、CMS垃圾回收器：使用并行的mark-sweep垃圾回收機制，被推薦使用，用來保持GC低開銷。雖然并行的GC會降低吞吐量，但是還是建議使用它，來減少batch的處理時間（降低處理過程中的gc開銷）。如果要使用，那么要在driver端和executor端都開啟。在spark-submit中使用–driver-java-options設置；使用spark.executor.extraJavaOptions參數(shù)設置。-XX:+UseConcMarkSweepGC。

些shuffle的父RDD的最大的分區(qū)數(shù)據(jù)。對于parallelize或者textFile這些輸入算子，因為沒有父RDD，所以依賴于ClusterManager的配置。如果是local模式，該默認值是local[x]中的x；如果是mesos的細粒度模式，該值為8，其它模式就是Math.max(2, 所有的excutor上的所有的core的總數(shù))。

7.5.3.序列化調優(yōu)說明

7.5.4. batchInterval

7.5.5. 內存調優(yōu)

總結

以上是生活随笔為你收集整理的SparkStreaming编程的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內容還不錯，歡迎將生活随笔推薦給好友。

SparkStreaming

上一篇： 2019.9.11 小米笔试算法岗编程题
下一篇：生活因态度而精彩