當(dāng)前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

SparkStreaming编程

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

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

0. SparkStreaming

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

1. 流式計(jì)算簡介

1.1. 流式計(jì)算

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

數(shù)據(jù)是無界的(unbounded)
數(shù)據(jù)是動(dòng)態(tài)的
計(jì)算速度是非?？斓?/p>
計(jì)算不止一次
計(jì)算不能終止

? 反過來看看一下離線計(jì)算有哪些特點(diǎn)：

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

1.2. 常見的離線和流式計(jì)算框架

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

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

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

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

1.3. SparkStreaming簡介

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

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

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

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

1.4. SparkStreaming基本工作原理

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

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

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

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

三者對(duì)比

storm和flink簡介

storm: storm.apache.org

flink: flink.apache.org

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

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

2. SparkStreaming實(shí)時(shí)處理入門案例

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

指定maven左表

執(zhí)行存儲(chǔ)位置

導(dǎo)入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的初始化，需要至少兩個(gè)參數(shù)，SparkConf和BatchDurationSparkConf不用多說batchDuration：提交兩次作業(yè)之間的時(shí)間間隔，每次會(huì)提交一個(gè)DStream，將數(shù)據(jù)轉(zhuǎn)化batch--->RDD所以說：sparkStreaming的計(jì)算，就是每隔多長時(shí)間計(jì)算一次數(shù)據(jù)*/val conf = new SparkConf().setAppName("SparkStreamingWordCount").setMaster("local[*]")val duration = Seconds(2)val ssc = new StreamingContext(conf, duration)//業(yè)務(wù)//為了執(zhí)行的流式計(jì)算，必須要調(diào)用start來啟動(dòng)ssc.start()//為了不至于start啟動(dòng)程序結(jié)束，必須要調(diào)用awaitTermination方法等待程序業(yè)務(wù)完成之后調(diào)用stop方法結(jié)束程序，或者異常ssc.awaitTermination()} }

2.2.2. 業(yè)務(wù)編寫

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的初始化，需要至少兩個(gè)參數(shù)，SparkConf和BatchDurationSparkConf不用多說batchDuration：提交兩次作業(yè)之間的時(shí)間間隔，每次會(huì)提交一個(gè)DStream，將數(shù)據(jù)轉(zhuǎn)化batch--->RDD所以說：sparkStreaming的計(jì)算，就是每隔多長時(shí)間計(jì)算一次數(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í)行的流式計(jì)算，必須要調(diào)用start來啟動(dòng)ssc.start()//為了不至于start啟動(dòng)程序結(jié)束，必須要調(diào)用awaitTermination方法等待程序業(yè)務(wù)完成之后調(diào)用stop方法結(jié)束程序，或者異常ssc.awaitTermination()} }

2.2.3. 使用netcat進(jìn)行測試

2.3. StreamingContext和Receiver說明

2.3.1. StreamingContext

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

關(guān)于local說明

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

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

start

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

- awaitTermination

2.3.2. Receiver

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

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

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

3. SparkStreaming和HDFS整合

3.1. 說明

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

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

監(jiān)聽本地

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

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

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

讀取文件的這種方式，沒有額外的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è)務(wù)隔離操作的消息隊(duì)列，而sparkstreaming是一款準(zhǔn)實(shí)時(shí)流式計(jì)算框架，所以二者的整合，是大勢所趨。

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

? 在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

導(dǎo)入依賴

<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來實(shí)現(xiàn)的。receiver從Kafka中獲取的數(shù)據(jù)都是存儲(chǔ)在Spark Executor的內(nèi)存中的，然后Spark Streaming啟動(dòng)的job會(huì)去處理那些數(shù)據(jù)。然而，在默認(rèn)的配置下，這種方式可能會(huì)因?yàn)榈讓拥氖《鴣G失數(shù)據(jù)。如果要啟用高可靠機(jī)制，讓數(shù)據(jù)零丟失，就必須啟用Spark Streaming的預(yù)寫日志機(jī)制（Write Ahead Log，WAL）。該機(jī)制會(huì)同步地將接收到的Kafka數(shù)據(jù)寫入分布式文件系統(tǒng)（比如HDFS）上的預(yù)寫日志中。所以，即使底層節(jié)點(diǎn)出現(xiàn)了失敗，也可以使用預(yù)寫日志中的數(shù)據(jù)進(jìn)行恢復(fù)。
需要注意的地方
Kafka的topic分區(qū)和Spark Streaming中生成的RDD分區(qū)沒有關(guān)系。在KafkaUtils.createStream中增加分區(qū)數(shù)量只會(huì)增加單個(gè)receiver的線程數(shù)，不會(huì)增加Spark的并行度
可以創(chuàng)建多個(gè)的Kafka的輸入DStream，使用不同的group和topic，使用多個(gè)receiver并行接收數(shù)據(jù)。
如果啟用了HDFS等有容錯(cuò)的存儲(chǔ)系統(tǒng)，并且啟用了寫入日志，則接收到的數(shù)據(jù)已經(jīng)被復(fù)制到日志中。因此，輸入流的存儲(chǔ)級(jí)別設(shè)置StorageLevel.MEMORY_AND_DISK_SER（即使用KafkaUtils.createStream（…，StorageLevel.MEMORY_AND_DISK_SER））的存儲(chǔ)級(jí)別。
數(shù)據(jù)會(huì)丟失原因

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()} }
說明
簡化的并行性：不需要?jiǎng)?chuàng)建多個(gè)輸入Kafka流并將其合并。使用directStream，Spark Streaming將創(chuàng)建與使用Kafka分區(qū)一樣多的RDD分區(qū)，這些分區(qū)將全部從Kafka并行讀取數(shù)據(jù)。所以在Kafka和RDD分區(qū)之間有一對(duì)一的映射關(guān)系。
效率：在第一種方法中實(shí)現(xiàn)零數(shù)據(jù)丟失需要將數(shù)據(jù)存儲(chǔ)在預(yù)寫日志中，這會(huì)進(jìn)一步復(fù)制數(shù)據(jù)。這實(shí)際
上是效率低下的，因?yàn)閿?shù)據(jù)被有效地復(fù)制了兩次:一次是Kafka，另一次是由預(yù)先寫入日志（Write
Ahead Log）復(fù)制。這個(gè)第二種方法消除了這個(gè)問題，因?yàn)闆]有接收器，因此不需要預(yù)先寫入日志。
只要Kafka數(shù)據(jù)保留時(shí)間足夠長。
正好一次（Exactly-once）的語義：第一種方法使用Kafka的高級(jí)API來在Zookeeper中存儲(chǔ)消耗的偏移量。傳統(tǒng)上這是從Kafka消費(fèi)數(shù)據(jù)的方式。雖然這種方法（結(jié)合提前寫入日志）可以確保零數(shù)據(jù)丟失（即至少一次語義），但是在某些失敗情況下，有一些記錄可能會(huì)消費(fèi)兩次。發(fā)生這種情況是因?yàn)镾park Streaming可靠接收到的數(shù)據(jù)與Zookeeper跟蹤的偏移之間的不一致。因此，在第二種方法中，我們使用不使用Zookeeper的簡單Kafka API。在其檢查點(diǎn)內(nèi)，Spark Streaming跟蹤偏移量。這消除了Spark Streaming和Zookeeper/Kafka之間的不一致，因此Spark Streaming每次記錄都會(huì)在發(fā)生故障的情況下有效地收到一次。為了實(shí)現(xiàn)輸出結(jié)果的一次語義，將數(shù)據(jù)保存到外部數(shù)據(jù)存儲(chǔ)區(qū)的輸出操作必須是冪等的，或者是保存結(jié)果和偏移量的原子事務(wù)。

冪等說明：多次操作結(jié)果都一樣，把這種操作稱之為冪等操作，比如數(shù)據(jù)庫的delete操作，或者：
INSERT INTO USER (id, NAME, age) VALUES(3, 'wangwu1', 16) ON DUPLICATE KEY UPDATE age=16, NAME='wangwu1'
offset的問題
/*offset的checkpoint(檢查點(diǎn))把需要管理的相關(guān)數(shù)據(jù)保存在某一個(gè)目錄下面，后續(xù)的時(shí)候直接從該目錄中讀取即可，在此處就是保存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)建或者恢復(fù)出來一個(gè)StreamingContextval ssc = StreamingContext.getOrCreate(checkpoint, createFunc)ssc.start()ssc.awaitTermination()} }
說明，這種方式，雖然能夠解決offset跟蹤問題，但是會(huì)在checkpoint目錄下面產(chǎn)生大量的小文件，并且操作進(jìn)行磁盤的IO操作，性能相對(duì)較差，那我們?nèi)绾喂芾韔ffset偏移量呢？

常見的管理offset偏移量的方式有如下：zookeeper、redis、hbase、mysql、elasticsearch、kafka，zookeeper也不建議使用，zookeeper太重要了，zk負(fù)載過高，容易出故障。
zookeeper管理offset
/*基于direct方式整合kafka使用zk手動(dòng)管理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中讀取手動(dòng)保存offset信息，然后從kafka指定offset位置開始讀取數(shù)據(jù)，當(dāng)然如果沒有讀取到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}"//判斷當(dāng)前路徑是否存在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

導(dǎo)入redis的maven依賴
<dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>2.9.0</version> </dependency>
提供redis的操作工具類(在spark-common模塊中處理)
//入口類：Jedis -->代表的是redis的一個(gè)客戶端連接，相當(dāng)于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>
代碼實(shí)現(xiàn)
/*基于direct方式整合kafka使用redis手動(dòng)管理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中讀取手動(dòng)保存offset信息，然后從kafka指定offset位置開始讀取數(shù)據(jù)，當(dāng)然如果沒有讀取到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ù)結(jié)構(gòu)：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ù)結(jié)構(gòu)

基于HBase進(jìn)行管理offset

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

在common模塊中完成hbaseconnection工具類的構(gòu)造
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對(duì)象 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消費(fèi)的編碼
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來來手動(dòng)管理offset信息保證數(shù)據(jù)被依次消費(fèi)1.手動(dòng)從hbase中讀取上一次消費(fèi)的offset信息有：從指定的offset位置開始消費(fèi)無：從offset為0或者最小的位置開始消費(fèi)2.使用指定offset從kafka中拉取數(shù)據(jù)3.拉取到數(shù)據(jù)之后進(jìn)行業(yè)務(wù)處理4.offset需要更新會(huì)指定的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就是多個(gè)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)//進(jìn)行各種業(yè)務(wù)邏輯操作//遍歷dstream中每一個(gè)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//將結(jié)果保存到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ù)結(jié)果：

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中讀取手動(dòng)保存offset信息，然后從kafka指定offset位置開始讀取數(shù)據(jù)，當(dāng)然如果沒有讀取到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}"//判斷當(dāng)前路徑是否存在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._/*** 事務(wù){(diào)* 保存數(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"// 設(shè)置jdbcClass.forName(driver)// 設(shè)置連接池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 開啟事務(wù)操作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ù)，并進(jìn)行處理；下一次再循環(huán)，如果批次的間隔時(shí)間為2s，但是數(shù)據(jù)的處理時(shí)間為3s，所以會(huì)有越來越多的沒有被處理的數(shù)據(jù)進(jìn)行累積，最后會(huì)拖垮程序，這不是我們所期望的。

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

spark.streaming.kafka.maxRatePerPartition: spark程序每秒中從每個(gè)partition分區(qū)讀取的最大的數(shù)據(jù)條數(shù)。比如batchInterval為2s，topic的分區(qū)為3，該參數(shù)的值為100，請(qǐng)問，每個(gè)批次最多可以讀取多少條數(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應(yīng)用案例

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

需求

數(shù)據(jù)格式：
<<<!>>>3111<<<!>>> --->記錄id <<<!>>>238<<<!>>> --->userid <<<!>>>20181111132902<<<!>>> --->用戶操作時(shí)間 <<<!>>>58.223.1.112<<<!>>> --->用戶ip地址 <<<!>>>202.102.92.18<<<!>>> --->用戶訪問ip地址 <<<!>>>59947<<<!>>> --->用戶端口 <<<!>>>80<<<!>>> --->服務(wù)端口 <<<!>>>www.sumecjob.com<<<!>>> -->服務(wù)域名 <<<!>>><<<!>>> <<<!>>><<<!>>> <<<!>>><<<!>>> <<<!>>><<<!>>> <<<!>>><<<!>>> <<<!>>>http://www.sumecjob.com/Social.aspx<<<!>>> -->用戶訪問的具體url <<<!>>>2556928066<<<!>>>
從kafka指定的topic中讀取如上數(shù)據(jù)，進(jìn)行清洗，剔除上述無用字段，保留有用信息，包括userid 用戶操作時(shí)間(timestamp) 用戶ip地址:端口服務(wù)地址：服務(wù)端口 url,最后將清洗結(jié)果送回kafka指定的topic中，完成在線etl。
topic構(gòu)建
--原始數(shù)據(jù)的topic kafka-topics.sh --create \ --topic orginal \ --zookeeper bigdata01:2181/kafka \ --partitions 3 --replication-factor 1--標(biāo)準(zhǔn)數(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()}})}//這種操作效率太低，會(huì)創(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ù)結(jié)構(gòu)：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í)行過程中，可能會(huì)出現(xiàn)如下異常：![1565342525589](assets/1565342525589.png)原因是因?yàn)槭褂昧司€程非安全的SimpleDateFormat造成的，如何解決！以如下案例來解決：```java /*演示SimpleDateFormat多線程安全問題執(zhí)行一下代碼，可能會(huì)出現(xiàn)NumberFormatException: For input string: "E.178E1" 主要原因是因?yàn)橛捎赟impleDateFormat為線程非安全的類所造成的*/

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”
);
//轉(zhuǎn)化為實(shí)踐戳，但是以多線程方式操作
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. 常見的算子操作

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

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是一個(gè)transformation算子，轉(zhuǎn)換算子。

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

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

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

5.2.2. 需求簡介——?jiǎng)討B(tài)黑名單過濾

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

5.2.3. 代碼實(shí)現(xiàn)

/*** 在線黑名單過濾** 類名起名規(guī)范* 首字母大寫，多單詞，采用駝峰* 一律名詞，不能動(dòng)詞* 并且單數(shù)不能復(fù)數(shù)* 方法名起名規(guī)范* 首字母小寫，多單詞，采用駝峰* 一般采用動(dòng)賓短語（動(dòng)詞+名詞）* 盡量少用一些漢語拼音，中文** 需求：* 從用戶請(qǐng)求的nginx日志中過濾出黑名單的數(shù)據(jù)，保留白名單數(shù)據(jù)進(jìn)行后續(xù)業(yè)務(wù)統(tǒng)計(jì)。* 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()} }

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

5.3. updateStateByKey

5.3.1. 概述

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

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

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

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

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

/*** 統(tǒng)計(jì)，截止到目前為止出現(xiàn)的每一個(gè)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:為當(dāng)前key的狀態(tài)option為key對(duì)應(yīng)的歷史值*/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))} }

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

5.4. window

5.4.1. 概述

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

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

這里面每一個(gè)time都是時(shí)間單元，在官方的例子中，每隔window size是3 time unit, 而且每隔2個(gè)單位時(shí)間，窗口會(huì)slide一次。
所以基于窗口的操作，需要指定2個(gè)參數(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).

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

滑動(dòng)間隔，就是我們可以理解的cron表達(dá)式吧。

5.4.2. 案例說明

/*** 統(tǒng)計(jì)，截止到目前為止出現(xiàn)的每一個(gè)key的次數(shù)* window窗口操作，每個(gè)多長M時(shí)間,通過過往N長時(shí)間內(nèi)產(chǎn)生的數(shù)據(jù)* M就是滑動(dòng)長度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:為當(dāng)前key的狀態(tài)option為key對(duì)應(yīng)的歷史值*/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最強(qiáng)大的地方在于，可以與Spark Core、Spark SQL整合使用，之前已經(jīng)通過transform、foreachRDD等算子看到，如何將DStream中的RDD使用Spark Core執(zhí)行批處理操作?，F(xiàn)在就來看看，如何將DStream中的RDD與Spark SQL結(jié)合起來使用。
案例：top3的商品排序：最新的top3

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

6.2. 代碼實(shí)現(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的緩存就只有一個(gè)方面，DStream對(duì)應(yīng)的RDD的緩存，RDD如何緩存？rdd.persist()，所以DStream的緩存說白了就是RDD的緩存，使用persist()指定，及其需要指定持久化策略，大多算子默認(rèn)情況下，持久化策略為MEMORY_AND_DISK_SER_2。

7.2. SparkStreaming的checkpoint機(jī)制

7.2.1 概述

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

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

7.2.2. 啟動(dòng)checkpoint

啟動(dòng)方式一

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

啟動(dòng)方式二

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

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

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

eg:

/*offset的checkpoint(檢查點(diǎn))把需要管理的相關(guān)數(shù)據(jù)保存在某一個(gè)目錄下面，后續(xù)的時(shí)候直接從該目錄中讀取即可，在此處就是保存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)建或者恢復(fù)出來一個(gè)StreamingContextval ssc = StreamingContext.getOrCreate(checkpoint, createFunc)ssc.start()ssc.awaitTermination()} }

? 當(dāng)程序?qū)?yīng)的driver失敗進(jìn)行恢復(fù)的時(shí)候，上述的修改，只是完成了第一步，還有第二步，第三步要走。

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

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

7.3. DriverHA

7.3.1. DriverHA的原理

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

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

恢復(fù)計(jì)算（圖2中的橙色箭頭）：使用Checkpoint數(shù)據(jù)重啟driver，重新構(gòu)造上下文并重啟接收器?；謴?fù)元數(shù)據(jù)塊（圖2中的綠色箭頭）：恢復(fù)Block元數(shù)據(jù)?；謴?fù)未完成的作業(yè)（圖2中的紅色箭頭）：使用恢復(fù)出來的元數(shù)據(jù)，再次產(chǎn)生RDD和對(duì)應(yīng)的job，然后提交到Spark集群執(zhí)行。通過如上的數(shù)據(jù)備份和恢復(fù)機(jī)制，Driver實(shí)現(xiàn)了故障后重啟、依然能恢復(fù)Streaming任務(wù)而不丟失數(shù)據(jù)，因此提供了系統(tǒng)級(jí)的數(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代碼實(shí)現(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程序的部署、升級(jí)與維護(hù)

部署

升級(jí)

監(jiān)控

7.5. 調(diào)優(yōu)建議

7.5.1.設(shè)置合理的CPU

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

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

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

7.5.2.關(guān)于接受數(shù)據(jù)的調(diào)優(yōu)說明

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

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

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

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

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

舉個(gè)例子

1)、batch interval為3s，block interval為150ms，會(huì)創(chuàng)建20個(gè)task。如果你認(rèn)為每個(gè)batch的task數(shù)量太少，即低于每臺(tái)機(jī)器的cpu core數(shù)量，那么就說明batch的task數(shù)量是不夠的，因?yàn)樗械腸pu資源無法完全被利用起來。要為batch增加block的數(shù)量，那么就減小block interval

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

7.5.2.設(shè)置合理的并行度

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

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

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

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

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

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

7.5.4. batchInterval

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

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

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

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

7.5.5. 內(nèi)存調(diào)優(yōu)

內(nèi)存調(diào)優(yōu)的另外一個(gè)方面是垃圾回收。對(duì)于流式應(yīng)用來說，如果要獲得低延遲，肯定不想要有因?yàn)镴VM垃圾回收導(dǎo)致的長時(shí)間延遲。有很多參數(shù)可以幫助降低內(nèi)存使用和GC開銷：

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

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

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

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

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

7.5.4. batchInterval

7.5.5. 內(nèi)存調(diào)優(yōu)

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

總結(jié)

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

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

SparkStreaming

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

3atv精品不卡视频,97人人超碰国产精品最新,中文字幕av一区二区三区人妻少妇,久久久精品波多野结衣,日韩一区二区三区精品

编程问答

SparkStreaming编程

0. SparkStreaming

1. 流式計(jì)算簡介

1.1. 流式計(jì)算

1.2. 常見的離線和流式計(jì)算框架

1.3. SparkStreaming簡介

1.4. SparkStreaming基本工作原理

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

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

2. SparkStreaming實(shí)時(shí)處理入門案例

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

2.2. 編碼

2.2.1. 入口類StreamingContext

2.2.2. 業(yè)務(wù)編寫

2.2.3. 使用netcat進(jìn)行測試

2.3. StreamingContext和Receiver說明

2.3.1. StreamingContext

2.3.2. Receiver

3. SparkStreaming和HDFS整合

3.1. 說明

3.2. 編碼

4. SparkStreaming與Kafka整合（*）

4.1. 整合簡述

4.2. spark-stremaing-kafka-0-8

4.2.1. Receiver的方式

4.2.2. Direct的方式

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

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

4.3.2. 常見問題

4.3.3. 在線ETL應(yīng)用案例

5. SparkStreaming常見transformation算子

5.1. 常見的算子操作

5.2. transform

5.2.1. 概述

5.2.2. 需求簡介——?jiǎng)討B(tài)黑名單過濾

5.2.3. 代碼實(shí)現(xiàn)

5.3. updateStateByKey

5.3.1. 概述

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

5.4. window

5.4.1. 概述

5.4.2. 案例說明

6. SparkSQL和SparkStreaming的整合案例

6.1. 需求

6.2. 代碼實(shí)現(xiàn)

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

7.1. SparkStreaming緩存操作

7.2. SparkStreaming的checkpoint機(jī)制

7.2.1 概述

7.2.2. 啟動(dòng)checkpoint

7.3. DriverHA

7.3.1. DriverHA的原理

7.3.2. DriverHA的配置

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

7.4. SparkStreaming程序的部署、升級(jí)與維護(hù)

7.5. 調(diào)優(yōu)建議

7.5.1.設(shè)置合理的CPU

7.5.2.關(guān)于接受數(shù)據(jù)的調(diào)優(yōu)說明

7.5.2.設(shè)置合理的并行度

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

7.5.4. batchInterval

7.5.5. 內(nèi)存調(diào)優(yōu)

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

7.5.4. batchInterval

7.5.5. 內(nèi)存調(diào)優(yōu)

總結(jié)