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

编程问答

SparkStreaming基础

發布時間：2024/7/5 编程问答 42 豆豆

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

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? SparkStreaming基礎

1?流式計算

1.1?常見的離線和流式計算框架

2?SparkStreaming簡介

2.1?核心概念DStream

2.2?工作原理

2.3?Storm，SparkStreaming和Flink的對比

2.4?如何選擇流式處理框架

3?SparkStreaming實時案例

3.1?StreamingContext和Receiver說明

4?SparkStreaming和HDFS整合

5?SparkStreaming與Kafka整合

5.1?spark-stremaing-kafka-0-8

6?消費kafka的時候常見的問題

6.1?生產速率過高

6.2?偏移量過期

7 DStream常見的的transformation

8?緩存和CheckPoint機制

8.1?緩存持久化

8.2?checkpoint機制

9?DriverHA

9.1 原理

9.2?DriverHA的配置

9.3?Driver代碼實現

10.?Spark Streaming 性能調優

10.1?設置合理的CPU

10.2?接受數據的調優

10.3?設置合理的并行度

10.4?序列化調優

10.5????????batchInterval

10.6?內存調優

???????

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? SparkStreaming基礎

1?流式計算

流式計算就像水流一樣，數據連綿不斷的產生，并被快速處理。

流式計算特點：①數據是無界的(unbounded)；②數據是動態的；③計算速度是非常快的；④計算不止一次；⑤計算不能終止

離線計算特點：①數據是有界的(Bounded) ；②數據靜態的；③計算速度通常較慢；④計算只執行一次；⑤計算終會終止

1.1?常見的離線和流式計算框架

常見的離線計算框架：mapreduce，spark-core，flink-dataset

常見的流式計算框架：storm(jstorm) ，spark-streaming，flink-datastream(blink)

2?SparkStreaming簡介

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

純實時的計算，指的是來一條記錄(event事件)，啟動一次計算的作業；離線計算，指的是每次計算一個非常大的一批(比如幾百G，好幾個T)數據；準實時呢，介于純實時和離線計算之間的一種計算方式。

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

DStream，本質上是RDD的序列。

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

2.1?核心概念DStream

類似于SparkCore中的RDD和SparkSQL中的Dataset、DataFrame，在SparkStreaming中的編程模型是DStream(離散化的流)。DStream是對一個時間段內產生的一些列RDD的封裝，也就是說一個DStream內部包含多個RDD。

DStream可以通過輸入算子來創建，也可以通過高階算子，比如map、flatMap等等進行轉換產生。

2.2?工作原理

對DStream應用的算子，其實在底層會被翻譯為對DStream中每個RDD的操作。比如對一個DStream執行一個map操作，會產生一個新的DStream。但是，在底層，對輸入DStream中每個時間段的RDD，都應用一遍map操作，然后生成的新的RDD，即作為新的DStream中的那個時間段的一個RDD。底層的RDD的transformation操作。

還是由Spark Core的計算引擎來實現的。Spark Streaming對Spark Core進行了一層封裝，隱藏了細節，然后對開發人員提供了方便易用的高層次的API。

2.3?Storm，SparkStreaming和Flink的對比

2.4?如何選擇流式處理框架

（1）Storm：

①建議在需要純實時，不能忍受1秒以上延遲的場景下使用，要求純實時進行交易和分析時。 ②要求可靠的事務機制和可靠性機制，即數據的處理完全精準，一條也不能多，一條也不能少，也可以考慮使用Storm，但是Spark Streaming也可以保證數據的不丟失。③如果我們需要考慮針對高峰低峰時間段，動態調整實時計算程序的并行度，以最大限度利用集群資源（通常是在小型公司，集群資源緊張的情況），我們也可以考慮用Storm

（2）Spark Streaming

①不滿足上述Storm要求的話，我們可以考慮使用Spark Streaming來進行實時計算。 ②考慮使用Spark Streaming最主要的一個因素，應該是針對整個項目進行宏觀的考慮，如果一個項目除了實時計算之外，還包括了離線批處理、交互式查詢、圖計算和MLIB機器學習等業務功能，而且實時計算中，可能還會牽扯到高延遲批處理、交互式查詢等功能，那么就應該首選Spark生態，用Spark Core開發離線批處理，用Spark SQL開發交互式查詢，用Spark Streaming開發實時計算，三者可以無縫整合，給系統提供非常高的可擴展性。

（3）Flink

支持高吞吐、低延遲、高性能的流處理支持帶有事件時間的窗口（Window）操作支持有狀態計算的Exactly-once語義支持高度靈活的窗口（Window）操作，支持基于time、count、session，以及data-driven的窗口操作支持具有Backpressure功能的持續流模型支持基于輕量級分布式快照（Snapshot）實現的容錯一個運行時同時支持Batch on Streaming處理和Streaming處理 Flink在JVM內部實現了自己的內存管理支持迭代計算支持程序自動優化：避免特定情況下Shuffle、排序等昂貴操作，中間結果有必要進行緩存。

3?SparkStreaming實時案例

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

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

3.1?StreamingContext和Receiver說明

（1）關于local

當將上述程序中的master由local[*],修改為local的時候，程序業務不變，發生只能接收數據，無法處理數據。 local[*]和local的區別：后者只為當前程序提供一個線程來處理，前者提供可用的所有的cpu的core來處理，當前情況下為2或者4。

SparkStreaming優先使用線程資源來接收數據，其次才是對數據的處理，接收數據的對象就是Receiver。所以，如果讀取數據的時候有receiver，程序的線程個數至少為2。

（2）關于start

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

start()方法只會讓當前的計算執行一次，要想持續不斷的進行接收數據，計算數據，就需要使用awaitTermination方法

（3）關于Receiver

Receiver就是數據的接收者，把資源分成了兩部分，一部分用來接收數據，一部分用來處理數據。Receiver接收到的數據，其實就是一個個的batch數據，是RDD，存儲在Executor內存。Receiver就是Executor內存中的一部分。

4?SparkStreaming和HDFS整合

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

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

（1）監聽本地：無法讀取手動拷貝，或者剪切到指定目錄下的文件，只能讀取通過流寫入的文件。

（2）監聽hdfs：正常情況下，我們可以讀取到通過put上傳的文件，還可以讀取通過cp拷貝的文件，但是讀取不了mv移動的文件。讀取文件的這種方式，沒有額外的Receiver消耗線程資源，所以可以指定master為local

object SparkStreamingHDFS {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[2]")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)//讀取local中數據 --->需要通過流的方式寫入 // val lines = ssc.textFileStream("file:///E:/data/monitored")//讀取hdfs中數據val lines = ssc.textFileStream("hdfs://bigdata01:9000/data/spark")lines.print()ssc.start()ssc.awaitTermination()} }

5?SparkStreaming與Kafka整合

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

SparkStreaming和Kafka的整合有兩大版本——spark-streaming-kafka-0-8和spark-streaming-kafka-0-10。spark-streaming-kafka-0-8版本還有兩種方式——Receiver和Direct方式。

5.1?spark-stremaing-kafka-0-8

依賴：

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

（1）Receiver的方式

/*** 使用kafka的receiver-api讀取數據*/ object SparkStreamingKafkaReceiverOps {def main(args: Array[String]): Unit = {val conf = new SparkConf().setAppName("SparkStreamingKafkaReceiverOps").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" -> "receiver","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來獲取數據。Receiver是使用Kafka的高層次Consumer API來實現的。receiver從Kafka中獲取的數據都是存儲在Spark Executor的內存中的，然后Spark Streaming啟動的job會去處理那些數據。

在默認的配置下，這種方式可能會因為底層的失敗而丟失數據。如果要啟用高可靠機制，讓數據零丟失，就必須啟用Spark Streaming的預寫日志機制（Write Ahead Log，WAL）。該機制會同步地將接收到的Kafka數據寫入分布式文件系統（比如HDFS）上的預寫日志中。所以，即使底層節點出現了失敗，也可以使用預寫日志中的數據進行恢復。（如果要開啟wal，需要在sparkconf中配置參數：spark.streaming.receiver.writeAheadLog.enable=true）

數據會丟失原因：

在上圖消費的過程中，需要處理數據和offset兩件事，任何一個出問題都會導致處理失敗：

①數據處理成功，保存偏移量失敗：數據被重復消費at-least-once

②偏移量保存成功，數據處理失敗：數據最多只能被處理一次at-most-once

③都處理成功：數據恰好處理一次：exactly-once

要想達到數據恰好處理一次那就只能將offset和數據處理保證在一個事務中，保證其原子性

注意：

①Kafka的topic分區和Spark Streaming中生成的RDD分區沒有關系。在KafkaUtils.createStream中增加分區數量只會增加單個receiver的線程數，不會增加Spark的并行度

②可以創建多個的Kafka的輸入DStream，使用不同的group和topic，使用多個receiver并行接收數據。

③如果啟用了HDFS等有容錯的存儲系統，并且啟用了寫入日志，則接收到的數據已經被復制到日志中。因此，輸入流的存儲級別設置StorageLevel.MEMORY_AND_DISK_SER（即使用KafkaUtils.createStream（...，StorageLevel.MEMORY_AND_DISK_SER））的存儲級別。

（2）Direct的方式

//基于direct方式整合kafka object parkStreamingKafkaDirectOps {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("parkStreamingKafkaDirectOps").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" -> "direct","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()} }

優勢：

①更加簡單的并行度：Direct模式下面，不再像receiver的方式，如果提高消費的性能，需要創建多個InputStream，最后將它們進行union，之后整體進行操作。Direct模式中，topic的分區數就和rdd的分區是一一對應的，此時我們每個rdd的分區就用來消費topic中的一個分區。

②性能較高：Receiver模式下面，為了保證數據零丟失，開啟了WAL。但是會對數據多兩次備份——kafka本身和wal，同時在備份過程中肯定需要時間，所以性能不是很高。而Direct模式，因為沒有receiver，也就沒有wal，只需要有足夠的retention策略，我們就可以從kafka中進行恢復。因為此時的偏移量被我們的程序所控制。

③一致性（Exactly-once）的語義

在流式計算過程中會有不同的語義，這些語義產生的原因，就是一條記錄被處理一次，被處理多次，一次或者一次也沒有被處理，對應語義稱之為Exactly once，at least once， at most once。不管是at least還是at most都會造成計算結果和真實的結果有偏差，不是我們所樂見的。

receiver的這種方式，會造成數據at least once，因為wal的存在，spark程序和偏移量之間的讀寫關系不一致。而在Direct的情況下可以保證數據的Exactly once semantics，因為我們使用的kafka底層的api，可以更加精準地在程序中把握偏移量。此時我們就不需要使用zookeeper，而使用checkpoint來存儲偏移量。

但是，為了保證數據輸出的一致性語義，則需要spark程序的輸出是冪等操作或者是原子性操作。

冪等說明：多次操作結果都一樣，把這種操作稱之為冪等操作，比如數據庫的delete操作

（3）offset的問題

/*offset的checkpoint(檢查點)把需要管理的相關數據保存在某一個目錄下面，后續的時候直接從該目錄中讀取即可，在此處就是保存offset數據*/ object CheckpointKafkaDirectOps {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("CheckpointKafkaDirectOps").setMaster("local")val duration = Seconds(2)val checkpoint = "file:///E:/data/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" -> "checkpointgroup","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}//創建或者恢復出來一個StreamingContextval ssc = StreamingContext.getOrCreate(checkpoint, createFunc)ssc.start()ssc.awaitTermination()} }

這種方式，雖然能夠解決offset跟蹤問題，但是會在checkpoint目錄下面產生大量的小文件，并且操作進行磁盤的IO操作，性能相對較差。那我們選擇其他方式管理offset偏移量，常見的管理offset偏移量的方式有如下：zookeeper、redis、hbase、mysql、elasticsearch、kafka，zookeeper也不建議使用，zookeeper太重要了，zk負載過高，容易出故障。

（4）zookeeper管理offset

/*基于direct方式整合kafka使用zk手動管理offset*/object KafkaDirectZKOps {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("KafkaDirectZKOps").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" -> "DirectZK","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位置開始讀取數據，如果沒有讀取到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交互的信息，比如數據的存放位置，存放格式官方的操作：/kafka/consumers/${group.id}/offsets/${topic}/${partition} -->data為offset存放位置&數據：/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} }

（5）redis管理offset

導入redis的maven依賴

<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>2.9.0</version> </dependency>

redis的操作工具類：

//入口類：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();} } /*基于direct方式整合kafka使用redis手動管理offset*/object KafkaDirectRedisOps {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("KafkaDirectRedisOps").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" -> "redisOps","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位置開始讀取數據，當然如果沒有讀取到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交互的信息，比如數據的存放位置，存放格式topic partition offset group五種數據結構：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.partition//untilOffset:結束offsetval offset = offsetRange.untilOffsetval gp = s"${group}|${partition}"jedis.hset(topic, gp, offset.toString)}JedisUtil.release(jedis)} }

6?消費kafka的時候常見的問題

6.1?生產速率過高

問題描述：SparkStreaming程序消費的能力要低于拉取kafka數據的能力，久而久之會造成數據的堆積，產生比如數據存儲壓力，計算壓力，造成程序異常。

（1）解決方式一：提高消費能力

提高消費能力，最簡單的就是提高并行度，在SparkStreaming基于Direct模式下增加topic的partition個數

（2）解決方式二：限流

添加一個配置參數即可——spark.streaming.kafka.maxRatePerPartition（基于Direct模式，每秒從每一個kafka 分區中讀取到的最大的記錄條數。

假設有一個Topic，分區有3個，streaming程序的batchInterval=2s,配置的改參數為spark.streaming.kafka.maxRatePerPartition=200,請問該stremaing程序每個批次最多能讀取：2 * 3 * 200 = 1200

6.2?偏移量過期

解決方案：每次從zk中讀到偏移量的之后，和當前Kafka topic partition的offset進行比較，如果異常，在讀取數據之前進行offset修正。如果zk中保存到的偏移量小于kafka讀取到的偏移量，將其設置為最小偏移量，反之設置為最大偏移量。

7 DStream常見的的transformation

map（func）	對DStream中的各個元素進行func函數操作，返回返回一個新的DStream
flatMap（func）	與map方法類似，不過各個輸入項可以被輸出為零個或多個輸出項
filter（func）	過濾出所有函數返回值為true的DStream元素并返回一個新的DStream
repartition（numPartition）	增加或減少DStream中的分區數，從而改變DStream的并行度
union（otherDStream）	將源DSTream和輸入參數為otherDStream的元素合并，返回一個新的DStream
count（）	通過對DStream中的各個RDD中的元素進行計數，然后返回只有一個元素的RDD構成一個DStream
reduce（func）	通過對DStream中的各個RDD中的元素利用func進行聚合操作，然后返回只有一個元素的RDD構成一個DStream
countByValue（）	對于元素類型為K的DStream，返回一個元素為（K，Long）鍵值對形式的新的DStream，Long對應的值為源DStream中各個RDD的key出現的次數
reduceByKey（func，[numTasks]）	利用func函數對源DStream中的key進行聚合操作，返回返回新的（K，V）對構成DStream
join（otherDStream，[numTasks]）	輸入為（K，V），（K，W）類型的DStream，返回一個新的（K，（V，W））類型的DStream
cogroup（otherDStream，[numTasks]）	輸入為（K，V），（K，W）類型的DStream，返回一個新的（K，（Seq[V],Seq[W]））類型的DStream
transform（func）	通過對RDD-to-RDD函數作用域源碼DStream中的各個RDD，可以是任意的RDD操作，返回一個新的RDD
updateStateByKey（func）	根據key的前置狀態和key的新志，對比key進行更新，返回一個新狀態的DStream
window函數	?

（1）cogroup：ogroup就是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)

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

最常見的就是DStream和rdd的join操作，還有DStream重分區(分區減少，coalsce)。也就是說transform主要就是用來自定義官方api沒有提供的一些操作。

下個舉個例子說明transform：動態黑名單過濾，利用SparkStreaming的流處理特性，可實現實時黑名單的過濾實現。可以使用leftouter join 對目標數據和黑名單數據進行關聯，將命中黑名單的數據過濾掉。

/*** 在線黑名單過濾** 需求：* 從用戶請求的nginx日志中過濾出黑名單的數據，保留白名單數據進行后續業務統計。* data structure* 27.19.74.143##2016-05-30 17:38:20##GET /static/image/common/faq.gif HTTP/1.1##200##1127*/ object lineBlacklistFilterOps {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("lineBlacklistFilterOps").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)))//接入外部的數據流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()//重分區 // filteredDStream.transform(_.coalesce(8))ssc.start()ssc.awaitTermination()} }

（3）updateStateByKey：根據于key的前置狀態和key的新值，對key進行更新，返回一個新狀態的Dstream。其實就是統計截止到目前為止key的值。

通過分析，在這個操作中需要兩個數據，一個是key的前置狀態，一個是key的新增(當前批次的數據)；還有歷史數據(前置狀態)得需要存儲在磁盤，不應該保存在內存中。同時key的前置狀態可能有可能沒有。

為了達到這個目的，最直接就得需要一個位置來存儲當前key的歷史的狀態。在SparkStreaming中，這個位置就存儲在checkpoint的目錄中。得需要兩個步驟：第一，定義狀態；第二，定義狀態函數。

舉例說明：計算截止到目前為止的全網總交易額，總流量

object SparkStreamingUSBOps {def main(args: Array[String]): Unit = {if(args == null || args.length < 3) {println("""|Parameter Errors ! Usage: <batchInterval> <host> <port>""".stripMargin)System.exit(-1)}val Array(batchInterval, host, port) = args //模式匹配val conf = new SparkConf().setAppName("SparkStreamingUSBOps").setMaster("local[*]")val checkpoint = "file:///E:/data/chk"def createFunc():StreamingContext = {val ssc = new StreamingContext(conf, Seconds(batchInterval.toLong))ssc.checkpoint(checkpoint)val linesDStream:ReceiverInputDStream[String] = ssc.socketTextStream(host, port.toInt)val pairs = linesDStream.flatMap(_.split("\\s+")).map((_, 1))//更新key的狀態val usb:DStream[(String, Int)] = pairs.updateStateByKey[Int]((seq, option) => updateFunc(seq, option))usb.print()ssc}/*** getActiveOrCreate: 獲的一個Active的StreamingContext或者創建一個信息* getOrCreate: 創建一個或者恢復一個*/val ssc = StreamingContext.getOrCreate(checkpoint, createFunc)ssc.start()ssc.awaitTermination()}/***基于key的前置狀態，和當前狀態進行合并，轉化為最新的狀態* 現在要統計總交易額：* 數據實時被打進計算系統。* long sum = 0;* 每一個批次的值：newValue* 總交易：sum += newValue* 前置狀態* 去年的時候，reduel的18歲，* 今年，refuel：19歲* @param current 當前批次的key對應的vlaue* @param previous 當前key對應的前置狀態，可能有，可能沒有* @return*/def updateFunc(current:Seq[Int], previous:Option[Int]): Option[Int] = {println(s"current: ${current.mkString(",")}--->previous: ${previous}") // var sum = 0 // for(cur <- current) { // sum += cur // } // val historyVal = previous.getOrElse(0) // sum += historyVal // // Option(sum)Option(current.sum + previous.getOrElse(0))} }

（4）window：窗口函數

是一個在流式計算領域中普遍都存在的一個概念——窗口函數。指的是一個窗口的數據，因為在Streaming中，每次計算的是一個批次的數據。這個窗口的概念跨域了批次，也就是說同時計算的數據，可以是多個批次的。

這個窗口操作，需要兩個參數：windowLength窗口長度；slidingInterval計算頻率或者滑動頻率。

每隔M長的時間，去統計過去N長時間產生的數據。M就是slidingInterval，N就是windowLength。

注意：此時不再以前的每個批次都提交一次作業，多個批次的數據合并到一起一同提交，所以需要擁有足夠的內存容納下下多個批次的數據。同時這里的streaming統計都是基于batchInterval來進行提交的，所以這里的windowLength和slidingInterval必須都是batchInterval（批處理時間間隔）的整數倍。

舉例說明如下：

/*** 基于window的窗口函數操作* 基本上所有的dstream的算子函數都有對應的window操作* 每隔2個時間單位，統計過去3個時間單位的數據* 1 2 3 4 5* ① ②*/ object parkStreamingWindowsOps {def main(args: Array[String]): Unit = {if(args == null || args.length < 3) {println("""|Parameter Errors ! Usage: <batchInterval> <host> <port>""".stripMargin)System.exit(-1)}val Array(batchInterval, host, port) = args //模式匹配val windowLen = Seconds(batchInterval.toLong * 3)val slidingDuration = Seconds(batchInterval.toLong * 2)val conf = new SparkConf().setAppName("parkStreamingWindowsOps").setMaster("local[*]")val ssc = new StreamingContext(conf, Seconds(batchInterval.toLong))val lines = ssc.socketTextStream(host, port.toInt)val ret = lines.flatMap(_.split("\\s+")).map((_, 1)).reduceByKeyAndWindow((v1:Int, v2:Int) => reduceFunc(v1, v2),windowLen,slidingDuration)ret.print()ssc.start()ssc.awaitTermination()}def reduceFunc(v1:Int, v2:Int):Int = v1 + v2 }

8?緩存和CheckPoint機制

8.1?緩存持久化

其實和RDD的持久化一樣，通過persist()方法來實現。需要指定持久化策略，大多算子默認情況下，持久化策略為MEMORY_AND_DISK_SER_2。

8.2?checkpoint機制

（1）為什么需要checkpoint

每一個Spark Streaming應用，正常來說，都是要7*24小時運轉的，這就是實時計算程序的特點。因為要持續不斷的對數據進行計算。因此，對實時計算應用的要求，應該是必須要能夠對與應用程序邏輯無關的失敗，進行容錯。

如果要實現這個目標，Spark Streaming程序就必須將足夠的信息checkpoint到容錯的存儲系統上，從而讓它能夠從失敗中進行恢復。

（2）Checkpoint相關

①???????元數據

配置信息：創建Spark Streaming應用程序的配置信息，比如SparkConf中的信息

DStream操作信息：定義了Spark Stream應用程序的計算邏輯的DStream操作信息。

未處理的batch信息：那些job正在排隊，還沒處理的batch信息。

②數據

對于一些將多個batch的數據進行聚合的，有狀態的transformation操作，這是非常有用的。在這種transformation操作中，生成的RDD是依賴于之前的batch的RDD的，這會導致隨著時間的推移，RDD的依賴鏈條變得越來越長。要避免由于依賴鏈條越來越長，導致的一起變得越來越長的失敗恢復時間，有狀態的transformation操作執行過程中間產生的RDD，會定期地被checkpoint到可靠的存儲系統上，比如HDFS。從而削減RDD的依賴鏈條，進而縮短失敗恢復時，RDD的恢復時間。

（3）???????啟動checkpoint

ssc.checkpoint(path)

注意：要注意的是，并不是說，所有的Spark Streaming應用程序，都要啟用checkpoint機制，如果即不強制要求從Driver失敗中自動進行恢復，又沒使用有狀態的transformation操作，那么就不需要啟用checkpoint。事實上，這么做反而是有助于提升性能的。

使用了有狀態的transformation操作——比如updateStateByKey，或者reduceByKeyAndWindow操作，被使用了，那么checkpoint目錄要求是必須提供的，也就是必須開啟checkpoint機制，從而進行周期性的RDD checkpoint。?

①開啟的方式

普通的checkpoint就使用上述的checkpoint即可，但是如果是driver要從失敗中進行恢復，就行修改程序。主要修改的就是StreamingContext的構建方式val ssc = StreamingContext.getOrCreate(checkpoint, func)

/*offset的checkpoint(檢查點)把需要管理的相關數據保存在某一個目錄下面，后續的時候直接從該目錄中讀取即可，在此處就是保存offset數據*/ object CheckpointWithKafkaDirectOps {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/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" -> "Checkpoint","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}//創建或者恢復出來一個StreamingContextval ssc = StreamingContext.getOrCreate(checkpoint, createFunc)ssc.start()ssc.awaitTermination()} }

9?DriverHA

9.1 原理

由于流計算系統是長期運行、且不斷有數據流入，因此其Spark守護進程（Driver）的可靠性至關重要，它決定了Streaming程序能否一直正確地運行下去。 Driver實現HA的解決方案就是將元數據持久化，以便重啟后的狀態恢復。如圖一所示，Driver持久化的元數據包括： Block元數據（圖中的綠色箭頭）：Receiver從網絡上接收到的數據，組裝成Block后產生的Block元數據； Checkpoint數據（圖中的橙色箭頭）：包括配置項、DStream操作、未完成的Batch狀態、和生成的RDD數據等；

恢復計算（圖中的橙色箭頭）：使用Checkpoint數據重啟driver，重新構造上下文并重啟接收器。恢復元數據塊（圖中的綠色箭頭）：恢復Block元數據。

恢復未完成的作業（圖中的紅色箭頭）：使用恢復出來的元數據，再次產生RDD和對應的job，然后提交到Spark集群執行。通過如上的數據備份和恢復機制，Driver實現了故障后重啟、依然能恢復Streaming任務而不丟失數據，因此提供了系統級的數據高可靠。

9.2?DriverHA的配置

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

9.3?Driver代碼實現

object SparkStreamingDriverHAOps {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()} }

10.?Spark Streaming 性能調優

10.1????????設置合理的CPU

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

（1）用于接收數據；

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

10.2?接受數據的調優

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

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

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

（3）使用inputStream.repartition(<number of partitions>)即可。這樣就可以將接收到的batch，分布到指定數量的機器上，然后再進行進一步的操作。

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

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

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

10.3?設置合理的并行度

如果在計算的任何stage中使用的并行task的數量沒有足夠多，那么集群資源是無法被充分利用的。舉例來說，對于分布式的reduce操作，比如reduceByKey和reduceByKeyAndWindow，默認的并行task的數量是由spark.default.parallelism參數決定的。你可以在reduceByKey等操作中，傳入第二個參數，手動指定該操作的并行度，也可以調節全局的spark.default.parallelism參數。該參數說的是，對于那些shuffle的父RDD的最大的分區數據。對于parallelize或者textFile這些輸入算子，因為沒有父RDD，所以依賴于ClusterManager的配置。如果是local模式，該默認值是local[x]中的x；如果是mesos的細粒度模式，該值為8，其它模式就是Math.max(2, 所有的excutor上的所有的core的總數)。

10.4?序列化調優

數據序列化造成的系統開銷可以由序列化格式的優化來減小。在流式計算的場景下，有兩種類型的數據需要序列化。

①輸入數據：默認情況下，接收到的輸入數據，是存儲在Executor的內存中的，使用的持久化級別是StorageLevel.MEMORY_AND_DISK_SER_2。這意味著，數據被序列化為字節從而減小GC開銷，并且會復制以進行executor失敗的容錯。因此，數據首先會存儲在內存中，然后在內存不足時會溢寫到磁盤上，從而為流式計算來保存所有需要的數據。這里的序列化有明顯的性能開銷——Receiver必須反序列化從網絡接收到的數據，然后再使用Spark的序列化格式序列化數據。

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

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

10.5????????batchInterval

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

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

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

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

10.6?內存調優

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

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

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

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

總結

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

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

上一篇： firewall mysql端口_Cen
下一篇： java 流程控制篇 2021/02/2