五、spark--sparkstreaming原理和使用

一、spark-streaming概述

1.1 常用的實(shí)時計(jì)算引擎

實(shí)時計(jì)算引擎也叫流式計(jì)算引擎，常用的目前有3個：
1、Apache Storm：真正的流式計(jì)算
2、Spark Streaming：嚴(yán)格上來說，不是真正的流式計(jì)算（實(shí)時計(jì)算）
? 把連續(xù)的流式數(shù)據(jù)，當(dāng)成不連續(xù)的RDD來處理
? 本質(zhì)：是一個離散計(jì)算（不連續(xù)的數(shù)據(jù)）
? 面試中問到時：先說它的本質(zhì)，
? 然后說自己的理解
? 常用的方法
? 和其他同類型技術(shù)的對比
3、Apache Flink：真正的流式計(jì)算。和Spark Streaming相反。
? 本質(zhì)：一個流式計(jì)算，雖然可以用于離線計(jì)算，但是本質(zhì)上是將離散數(shù)據(jù)模擬成流式數(shù)據(jù)來給flink做流式計(jì)算

創(chuàng)新互聯(lián)專注于橋西企業(yè)網(wǎng)站建設(shè),自適應(yīng)網(wǎng)站建設(shè),商城網(wǎng)站定制開發(fā)。橋西網(wǎng)站建設(shè)公司,為橋西等地區(qū)提供建站服務(wù)。全流程按需求定制開發(fā)，專業(yè)設(shè)計(jì)，全程項(xiàng)目跟蹤，創(chuàng)新互聯(lián)專業(yè)和態(tài)度為您提供的服務(wù)

1.2 spark-streaming是什么

? Spark Streaming是核心Spark API的擴(kuò)展，可實(shí)現(xiàn)可擴(kuò)展、高吞吐量、可容錯的實(shí)時數(shù)據(jù)流處理。數(shù)據(jù)可以從諸如Kafka，F(xiàn)lume，Kinesis或TCP套接字等眾多來源獲取，并且可以使用由高級函數(shù)（如map，reduce，join和window）開發(fā)的復(fù)雜算法進(jìn)行流數(shù)據(jù)處理。最后，處理后的數(shù)據(jù)可以被推送到文件系統(tǒng)，數(shù)據(jù)庫和實(shí)時儀表板。而且，您還可以在數(shù)據(jù)流上應(yīng)用Spark提供的機(jī)器學(xué)習(xí)和圖處理算法。

特點(diǎn)：
1、易用：集成在Spark中
2、容錯性：底層RDD，RDD本身就具備容錯機(jī)制。
3、支持多種編程語言：Java Scala Python

1.3 spark-streaming架構(gòu)

spark-streaming用來接收實(shí)時數(shù)據(jù)，然后處理程序通過類似于定時采樣的方式分批取得數(shù)據(jù)，每一批數(shù)據(jù)就是一個RDD，最終輸入給處理程序的是一個RDD隊(duì)列流，這個流其實(shí)就是discretizedstream或DStream。在內(nèi)部，DStream 由一個RDD序列表示。DStream對象就是可以用來調(diào)用各種算子進(jìn)行處理
五、spark--spark streaming原理和使用

圖1.1 DStream原理

1.4 案例演示--NetworkWordCount

首先啟動netcat服務(wù)器，并監(jiān)聽在端口1234上

nc -l 1234

沒有這個命令就 yum -y install netcat 安裝一下

接著啟動spark-streaming樣例程序，從本地的1234端口獲取數(shù)據(jù)，并進(jìn)行wordcount操作

到spark的安裝目錄下，執(zhí)行bin目錄下的命令：
bin/run-example streaming.NetworkWordCount localhost 1234

然后在netcat端輸入各種字符串：

[root@bigdata121 hive-1.2.1-bin]# nc -l 1234
king king hello

在另外一個窗口查看統(tǒng)計(jì)信息：

-------------------------------------------
Time: 1567005584000 ms
-------------------------------------------
(hello,1)
(king,2)

這邊就立馬統(tǒng)計(jì)出來了

1.5 自行編寫NetworkWordCount

首先maven中pom.xml記得再加上streaming的依賴（為了方便最好spark所有組件的依賴都加上）
pom.xml

<?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">
    <modelVersion>4.0.0</modelVersion>

    <groupId>king</groupId>
    <artifactId>sparkTest</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <spark.version>2.1.0</spark.version>
        <scala.version>2.11.8</scala.version>
        <hadoop.version>2.7.3</hadoop.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
            <version>2.1.0</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.1.0</version>
            <!--<scope>provided</scope>-->
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_2.11</artifactId>
            <version>2.1.0</version>
            <scope>runtime</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka_2.11</artifactId>
            <version>1.6.3</version>
        </dependency>

        <dependency>
            <groupId>MySQL</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.12</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.hive/hive-jdbc -->
        <dependency>
            <groupId>org.apache.hive</groupId>
            <artifactId>hive-jdbc</artifactId>
            <version>1.2.1</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/log4j/log4j -->
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.flume/flume-ng-core -->
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-core</artifactId>
            <version>1.8.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flume/flume-ng-sdk -->
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-sdk</artifactId>
            <version>1.8.0</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.flume/flume-ng-configuration -->
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-configuration</artifactId>
            <version>1.8.0</version>
        </dependency>

    </dependencies>

    <!--下面這是maven打包scala的插件，一定要，否則直接忽略scala代碼-->
    <build>
        <plugins>

            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <version>2.15.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>2.19</version>
                <configuration>
                    <skip>true</skip>
                </configuration>
            </plugin>

        </plugins>
    </build>

</project>

代碼：

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * wordcount流式計(jì)算程序
  *
  * 1、創(chuàng)建streamingContext對象
  *    創(chuàng)建DStream流（離散流）
  *    本質(zhì)是離散計(jì)算
  *
  *    離散：將連續(xù)數(shù)據(jù)變成離散數(shù)據(jù)，并實(shí)時立刻處理
  *    離線：并非是實(shí)時處理的
  *
  * 2、DStream表現(xiàn)形式就是RDD
  *    和操作RDD一樣
  *
  * 3、使用DStream將連續(xù)的數(shù)據(jù)庫切割成離散的RDD
  */
object NetworkWordCount {
  def main(args: Array[String]): Unit = {
    //設(shè)置日志級別為ERROR，默認(rèn)是INFO
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
    //Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    /**
    這是 StreamingContext 對象的標(biāo)準(zhǔn)創(chuàng)建方式
    無法通過 sparkSession對象來創(chuàng)建
    */
    //創(chuàng)建streamingContext對象,指定master為local[2]，意思是使用至少兩個核心，即兩個線程，一個用于發(fā)送數(shù)據(jù)，一個處理數(shù)據(jù)
    val conf = new SparkConf().setAppName("streaming wordcount").setMaster("local[2]")
    //這里指定conf對象，還有批處理的時間間隔為3秒，每3秒切一個rdd，然后處理.
    val streamingContext = new StreamingContext(conf, Seconds(3))

    //創(chuàng)建接收數(shù)據(jù)源，這里創(chuàng)建socketstream，接收數(shù)據(jù)，內(nèi)部會自動切割成一個個rdd。
    //指定監(jiān)聽的主機(jī)端口
    val streamText = streamingContext.socketTextStream("bigdata121", 1234, StorageLevel.MEMORY_ONLY)

    //wordcount流程
    val rdd1 = streamText.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)

    //打印結(jié)構(gòu)
    rdd1.print()

    //啟動streamingContext,開始計(jì)算
    streamingContext.start()

    //等待任務(wù)結(jié)束
    streamingContext.awaitTermination()
  }

}

在bigdata121虛擬機(jī)上啟動netcat服務(wù)：

nc -l 1234

idea中運(yùn)行上面的程序，并在netcat中輸入字符，結(jié)構(gòu)和實(shí)例的一樣

二、streaming基本原理和使用

2.1 StreamingContext對象的概念

1、StreamingContext會內(nèi)在的創(chuàng)建一個SparkContext的實(shí)例（所有Spark功能的起始點(diǎn)），你可以通過ssc.sparkContext訪問到這個實(shí)例。

2、一旦一個StreamingContext開始運(yùn)作，就不能設(shè)置或添加新的流計(jì)算。

3、一旦一個上下文被停止，它將無法重新啟動。

4、同一時刻，一個JVM中只能有一個StreamingContext處于活動狀態(tài)。

5、StreamingContext上的stop()方法也會停止SparkContext。 要僅停止StreamingContext（保持SparkContext活躍），請將stop() 方法的可選參數(shù)stopSparkContext設(shè)置為false。

6、只要前一個StreamingContext在下一個StreamingContext被創(chuàng)建之前停止（不停止SparkContext），SparkContext就可以被重用來創(chuàng)建多個StreamingContext。

2.2 離散流(DStreams):Discretized Streams

? DStream對象可以說整個spark-streaming程序的一個數(shù)據(jù)的出口，處理的數(shù)據(jù)都從這里來。前面也說了，這個對象里面其實(shí)一個個的RDD，這是DStream的本質(zhì)。而且經(jīng)過算子的轉(zhuǎn)換之后，DStream仍舊是DStream對象，里面也還是RDD。所以算子轉(zhuǎn)換的過程和普通RDD的概率類似?？偟膩碚fstreaming程序中，就是DStream之間的轉(zhuǎn)換，本質(zhì)上就是DStream中的RDD的轉(zhuǎn)換

2.3 DStream的算子

算子列表：
五、spark--spark streaming原理和使用

圖2.1 DStream算子

和普通rdd很類似，有兩個比較特殊的算子，transform和updateStateByKey

2.3.1 transform

transform(RDD[T]=>RDD[U])
是一個用于將dstream中的rdd轉(zhuǎn)換成新的rdd的算子。所以要注意，這個算子中的處理函數(shù)是接收rdd作為參數(shù)，不像其他算子是接收rdd中的數(shù)據(jù)作為參數(shù)的。

例子：
    val conf = new SparkConf().setAppName("streaming wordcount").setMaster("local[2]")
    //這里指定conf對象，還有批處理的時間間隔為4秒，每4秒切一個rdd，然后處理.
    val streamingContext = new StreamingContext(conf, Seconds(3))

    //創(chuàng)建socketstream，接收數(shù)據(jù)，內(nèi)部會自動切割成一個個rdd
    val streamText = streamingContext.socketTextStream("bigdata121", 1234, StorageLevel.MEMORY_ONLY)

    //接收的函數(shù)參數(shù)中就是rdd，然后在里面對rdd進(jìn)行處理，最后返回新的rdd
    streamText.transform(rdd=>{
      rdd.flatMap(_.split(" "))
    })

2.3.2 updateStateByKey

? 默認(rèn)情況下，Spark Streaming 不記錄之前的狀態(tài)，每次發(fā)一條數(shù)據(jù)，都從0開始。比如說進(jìn)行單詞統(tǒng)計(jì)時，之前統(tǒng)計(jì)的單詞數(shù)量并不會累加到下一次的統(tǒng)計(jì)中，下一次是從0開始計(jì)數(shù)的。如果想進(jìn)行累加操作，使用這個算子來實(shí)現(xiàn)這個功能

updateStateByKey((Seq[T],Option[S])=>Option[S])
這個算子接收的函數(shù)的參數(shù)要求有兩個：
Seq[T]：當(dāng)前對key進(jìn)行分組后，同一個key的value的一個集合，比如（"age",[1,2,1,1]）中的[1,2,1,1]
Option[S]:同一個key，在此之前的value總和，也就是這個key之前的計(jì)數(shù)狀態(tài)
返回值是之前的計(jì)數(shù)+現(xiàn)在的計(jì)數(shù)的一個返回值

例子：
下面將之前的wordcount改變一些，實(shí)現(xiàn)單詞的持續(xù)計(jì)數(shù)，不會每次都重新從0開始計(jì)數(shù)

package SparkStreamExer

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 測試updateStateByKey 進(jìn)行狀態(tài)的累加
  */
object TestUpdateState {
  def main(args: Array[String]): Unit = {
    //設(shè)置日志級別為ERROR，默認(rèn)是INFO
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    //創(chuàng)建streamingContext對象,指定master為local[2]，意思是使用至少兩個核心，即兩個線程，一個用于發(fā)送數(shù)據(jù)，一個處理數(shù)據(jù)
    val conf = new SparkConf().setAppName("streaming wordcount").setMaster("local[2]")
    //這里指定conf對象，還有批處理的時間間隔為4秒，每4秒切一個rdd，然后處理.
    val streamingContext = new StreamingContext(conf, Seconds(3))

    //設(shè)置檢查點(diǎn)，保存之前狀態(tài),需要保證目錄不存在
    streamingContext.checkpoint("hdfs://bigdata121:9000/sparkCheckpoint/spark-streaming")

    //創(chuàng)建socketstream，接收數(shù)據(jù)，內(nèi)部會自動切割成一個個rdd
    val streamText = streamingContext.socketTextStream("bigdata121", 1234, StorageLevel.MEMORY_ONLY)

    //切割數(shù)據(jù)，并添加計(jì)數(shù)對
    val wordPair = streamText.flatMap(_.split(" ")).map((_,1))

    //累加處理函數(shù)
    val addFunc = (currentValues:Seq[Int], previousValue:Option[Int]) => {
      //當(dāng)前值累加
      val currentSum = currentValues.sum

      //取出之前的值.如果值不存在就返回0
      val pre = previousValue.getOrElse(0)

      //之前和現(xiàn)在的值相加
      Option(pre + currentSum)
    }

    //更新，將舊計(jì)數(shù)更新為新計(jì)數(shù)狀態(tài)
    wordPair.updateStateByKey(addFunc).print()

    //啟動streamingContext,開始計(jì)算
    streamingContext.start()

    //等待任務(wù)結(jié)束
    streamingContext.awaitTermination()
  }
}

運(yùn)行這個demo的過程出現(xiàn)的報(bào)錯：

Caused by: java.lang.ClassNotFoundException: org.apache.commons.io.Charsets

說是沒有org.apache.commons.io.Charsets 這個類，進(jìn)去org.apache.commons.io看了下，果然沒有，估計(jì)是包版本太舊了，沒有這個類，百度了一下，2.5版本的有這個類，所以就在pom.xml添加上新的依賴

<dependency>
    <groupId>commons-io</groupId>
    <artifactId>commons-io</artifactId>
    <version>2.5</version>
</dependency>

接著運(yùn)行，OK了

2.3.3 foreachRDD

這個算子類似forech，但是操作的對象是整個rdd，不是rdd中的某些數(shù)據(jù)。

foreachRDD(RDD[T]=>Unit)
一般用于將rdd的結(jié)果寫入其他存儲中，比如hdfs，mysql等

下面有一個關(guān)于 foreachRDD和sql 的例子。

2.4 窗口操作

應(yīng)用場景：
一般用于統(tǒng)計(jì)最近N小時的數(shù)據(jù)，這樣的應(yīng)用的場景，這時候就需要窗口

2.4.1 原理

原理圖：
五、spark--spark streaming原理和使用

圖2.2 spark-streaming窗口操作

? 窗口其實(shí)就是DStream的基礎(chǔ)上，再加上一個時間范圍。如圖所示，每當(dāng)窗口滑過originalDStream時，落在窗口內(nèi)的源RDD被組合并被執(zhí)行操作以產(chǎn)生windowed DStream的RDD。在上面的例子中，操作應(yīng)用于最近3個時間單位的數(shù)據(jù)，并以2個時間單位滑動。所以窗口操作比起普通的DStream操作，普通的DStream是一個個RDD處理，而窗口則是一個時間范圍內(nèi)的RDD一起處理。而且窗口是DStream再上一層的一個封裝。
? 使用窗口的時候，有兩個關(guān)鍵參數(shù)：
窗口長度（windowlength）：窗口的時間長度（上圖的示例中為：3）
滑動間隔（slidinginterval）：兩次相鄰的窗口操作的間隔（即每次滑動的時間長度）（上圖示例中為：2）
而且要注意的一點(diǎn)是：這兩個參數(shù)必須是源DStream的采樣間隔的倍數(shù)（上圖示例中為：1）。因?yàn)槿绻皇钦麛?shù)倍，就會導(dǎo)致窗口邊緣會將一個rdd分隔成兩份，這樣是不行的，spark沒辦法處理半個rdd，rdd是不可分的。

2.4.2 窗口操作的相關(guān)算子

window(windowLength, slideInterval)
->基于源DStream產(chǎn)生的窗口化的批數(shù)據(jù)計(jì)算一個新的DStream

countByWindow(windowLength, slideInterval)
->返回流中元素的一個滑動窗口數(shù)

reduceByWindow(func, windowLength, slideInterval)
->返回一個單元素流。利用函數(shù)func聚集滑動時間間隔的流的元素創(chuàng)建這個單元素流。函數(shù)必須是相關(guān)聯(lián)的以使計(jì)算能夠正確的并行計(jì)算。

reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])
->應(yīng)用到一個(K,V)對組成的DStream上，返回一個由(K,V)對組成的新的DStream。每一個key的值均由給定的reduce函數(shù)聚集起來。注意：在默認(rèn)情況下，這個算子利用了Spark默認(rèn)的并發(fā)任務(wù)數(shù)去分組。你可以用numTasks參數(shù)設(shè)置不同的任務(wù)數(shù)

reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])
->上述reduceByKeyAndWindow() 的更高效的版本，其中使用前一窗口的reduce計(jì)算結(jié)果遞增地計(jì)算每個窗口的reduce值。這是通過對進(jìn)入滑動窗口的新數(shù)據(jù)進(jìn)行reduce操作，以及“逆減（inverse reducing）”離開窗口的舊數(shù)據(jù)來完成的。一個例子是當(dāng)窗口滑動時對鍵對應(yīng)的值進(jìn)行“一加一減”操作。但是，它僅適用于“可逆減函數(shù)（invertible reduce functions）”，即具有相應(yīng)“反減”功能的減函數(shù)（作為參數(shù)invFunc）。 像reduceByKeyAndWindow一樣，通過可選參數(shù)可以配置reduce任務(wù)的數(shù)量。 請注意，使用此操作必須啟用檢查點(diǎn)。

countByValueAndWindow(windowLength, slideInterval, [numTasks])
->應(yīng)用到一個(K,V)對組成的DStream上，返回一個由(K,V)對組成的新的DStream。每個key的值都是它們在滑動窗口中出現(xiàn)的頻率。

比較常用的是reduceByKeyAndWindow這個，常用于統(tǒng)計(jì)固定最近一段時間內(nèi)的數(shù)據(jù)，比如統(tǒng)計(jì)最近1小時訂單銷售量。下面把這個算子應(yīng)用到wordcount例子中。

2.4.3 例子

窗口大小為30s，每10s滑動一次窗口，并且對單詞的計(jì)數(shù)是累加的

package SparkStreamExer

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 測試updateStateByKey 進(jìn)行狀態(tài)的累加
  */
object TestUpdateState {
  def main(args: Array[String]): Unit = {
    //設(shè)置日志級別為ERROR，默認(rèn)是INFO
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    //創(chuàng)建streamingContext對象,指定master為local[2]，意思是使用至少兩個核心，即兩個線程，一個用于發(fā)送數(shù)據(jù)，一個處理數(shù)據(jù)
    val conf = new SparkConf().setAppName("streaming wordcount").setMaster("local[2]")
    //這里指定conf對象，還有批處理的時間間隔為4秒，每4秒切一個rdd，然后處理.
    val streamingContext = new StreamingContext(conf, Seconds(1))

    //設(shè)置檢查點(diǎn)，保存之前狀態(tài),需要保證目錄不存在
    streamingContext.checkpoint("hdfs://bigdata121:9000/sparkCheckpoint/spark-streaming3")

    //創(chuàng)建socketstream，接收數(shù)據(jù)，內(nèi)部會自動切割成一個個rdd
    val streamText = streamingContext.socketTextStream("bigdata121", 1234, StorageLevel.MEMORY_ONLY)

    //切割數(shù)據(jù)，并添加計(jì)數(shù)對
    val wordPair = streamText.flatMap(_.split(" ")).map((_,1))

    //在這里添加一個窗口操作
    val windowValue = wordPair.reduceByKeyAndWindow((x:Int,y:Int)=>x+y, Seconds(30), Seconds(10))

    //累加處理函數(shù)
    val addFunc = (currentValues:Seq[Int], previousValue:Option[Int]) => {
      //當(dāng)前值累加
      val currentSum = currentValues.sum

      //取出之前的值.如果值不存在就返回0
      val pre = previousValue.getOrElse(0)

      //之前和現(xiàn)在的值相加
      Option(pre + currentSum)
    }

    //更新，將舊計(jì)數(shù)更新為新計(jì)數(shù)狀態(tài)
    //wordPair.updateStateByKey(addFunc).print()
    windowValue.updateStateByKey(addFunc).print()

    //啟動streamingContext,開始計(jì)算
    streamingContext.start()

    //等待任務(wù)結(jié)束
    streamingContext.awaitTermination()
  }
}

2.5 sql操作

package SparkStreamExer

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 將streaming的DStream轉(zhuǎn)為可以使用sql操作
  */
object StreamingAndSql {
  def main(args: Array[String]): Unit = {
    //設(shè)置日志級別為ERROR，默認(rèn)是INFO
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    val conf = new SparkConf().setAppName("streaming and sql").setMaster("local[2]")
    val ssc = new StreamingContext(conf, Seconds(2))

    val lines = ssc.socketTextStream("bigdata121",1234, StorageLevel.MEMORY_ONLY)

    val words = lines.flatMap(_.split(" "))

    //需要將rdd轉(zhuǎn)為df對象，才能用于spark sql操作
    words.foreachRDD(rdd => {
      //從rdd中獲取conf配置，保證配置和rdd的配置一樣
      val spark = SparkSession.builder().config(rdd.sparkContext.getConf).getOrCreate()
      import spark.sqlContext.implicits._
      //rdd轉(zhuǎn)為df，并指定列名
      val df = rdd.toDF("word")
      //創(chuàng)建視圖并執(zhí)行sql
      df.createOrReplaceTempView("tmp1")
      val resultDF = spark.sql("select word,count(1) from tmp1 group by word")
      resultDF.show()
    })

    ssc.start()
    ssc.awaitTermination()

  }

}

2.6 checkpoint檢查點(diǎn)

這個和rdd中類似，只不過streaming中是通過 StreamingContext對象進(jìn)行checkpoint：

//創(chuàng)建streamingContext對象,指定master為local[2]，意思是使用至少兩個核心，即兩個線程，一個用于發(fā)送數(shù)據(jù)，一個處理數(shù)據(jù)
    val conf = new SparkConf().setAppName("streaming wordcount").setMaster("local[2]")
    //這里指定conf對象，還有批處理的時間間隔為4秒，每4秒切一個rdd，然后處理.
    val streamingContext = new StreamingContext(conf, Seconds(1))

    //設(shè)置檢查點(diǎn)，保存之前狀態(tài),需要保證目錄不存在
    streamingContext.checkpoint("hdfs://bigdata121:9000/sparkCheckpoint/spark-streaming3")

三、streaming的數(shù)據(jù)源

3.1 基本數(shù)據(jù)源

文件流：textFileStream
套接字流：socketTextStream/sockeStream，前面已經(jīng)講過例子，這里不重復(fù)
RDD隊(duì)列流：queueStream

1、textFileStream
通過監(jiān)控文件系統(tǒng)的變化，若有新文件添加，則將它讀入并作為數(shù)據(jù)流
需要注意的是：
這些文件具有相同的格式
這些文件通過原子移動或重命名文件的方式在dataDirectory創(chuàng)建
如果在文件中追加內(nèi)容，這些追加的新數(shù)據(jù)也不會被讀取。

例子：
package SparkStreamExer

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object StreamingFromFile {
  def main(args: Array[String]): Unit = {
    //設(shè)置日志級別為ERROR，默認(rèn)是INFO
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    val conf = new SparkConf().setAppName("spark window operation").setMaster("local[2]")

    val ssc = new StreamingContext(conf, Seconds(4))

    val fileStream = ssc.textFileStream("G:\\test\\teststreaming")

    fileStream.print()

    ssc.start()
    ssc.awaitTermination()
  }
}

//==========================================================
2、queueStream
RDD隊(duì)列流是從一個隊(duì)列中讀取RDD
例子：
package SparkStreamExer

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable

object StreamingFromRDDQueue {
  def main(args: Array[String]): Unit = {
    //設(shè)置日志級別為ERROR，默認(rèn)是INFO
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    val conf = new SparkConf().setAppName("spark streaming rdd queue").setMaster("local[2]")

    val ssc = new StreamingContext(conf, Seconds(4))

    //創(chuàng)建隊(duì)列
    val rddQueue = new mutable.Queue[RDD[Int]]()

    //隊(duì)列中添加rdd
    for (x<- 1 to 3) {
      rddQueue += ssc.sparkContext.makeRDD(1 to 10)
    }

    //從隊(duì)列讀取rdd
    val queueRdd = ssc.queueStream(rddQueue).map(_*2)
    queueRdd.print()

    ssc.start()
    ssc.awaitTermination()
  }
}

3.2 高級數(shù)據(jù)源

高級數(shù)據(jù)源一般在生產(chǎn)中比較常用，很少使用spark直接監(jiān)控?cái)?shù)據(jù)的。常用的高級數(shù)據(jù)源有Kafka,Flume,Kinesis,Twitter等等。下面主要講解flume

3.2.1flume

1、flume推送數(shù)據(jù)到計(jì)算節(jié)點(diǎn)
（1）首先配置flume的agent配置文件

a1.sources=r1
a1.channels=c1
a1.sinks=k1

# 監(jiān)控目錄
a1.sources.r1.type=TAILDIR
a1.sources.r1.filegroups=f1
a1.sources.r1.filegroups.f1=/opt/modules/apache-flume-1.8.0-bin/logs/.*
a1.sources.r1.fileHeader=true

a1.channels.c1.type=memory
a1.channels.c1.capacity=10000
a1.channels.c1.transactionCapacity=100

# 我這里是在ide中直接運(yùn)行spark程序，所以flume數(shù)據(jù)直接推導(dǎo)windows主機(jī)上
a1.sinks.k1.type=avro
a1.sinks.k1.hostname=192.168.50.1
a1.sinks.k1.port=1234

a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1

（2）spark代碼
pom.xml

<?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">
    <modelVersion>4.0.0</modelVersion>

    <groupId>SparkDemo</groupId>
    <artifactId>SparkDemoTest</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <spark.version>2.1.0</spark.version>
        <scala.version>2.11.8</scala.version>
        <hadoop.version>2.7.3</hadoop.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
            <version>2.1.0</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>2.1.0</version>
            <!--<scope>provided</scope>-->
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_2.11</artifactId>
            <version>2.1.0</version>
            <scope>runtime</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka_2.11</artifactId>
            <version>1.6.3</version>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.12</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.hive/hive-jdbc -->
        <dependency>
            <groupId>org.apache.hive</groupId>
            <artifactId>hive-jdbc</artifactId>
            <version>1.2.1</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/log4j/log4j -->
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.flume/flume-ng-core -->
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-core</artifactId>
            <version>1.8.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.flume/flume-ng-sdk -->
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-sdk</artifactId>
            <version>1.8.0</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.flume/flume-ng-configuration -->
        <dependency>
            <groupId>org.apache.flume</groupId>
            <artifactId>flume-ng-configuration</artifactId>
            <version>1.8.0</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-flume -->

        <!--這里是spark從flume讀取數(shù)據(jù)的依賴，不要忘了-->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-flume_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-flume-sink -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-flume-sink_2.11</artifactId>
            <version>2.1.0</version>
        </dependency>

        <dependency>
            <groupId>commons-io</groupId>
            <artifactId>commons-io</artifactId>
            <version>2.5</version>
        </dependency>

    </dependencies>

    <!--下面這是maven打包scala的插件，一定要，否則直接忽略scala代碼-->
    <build>
        <plugins>

            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <version>2.15.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <plugin>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.6.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>2.19</version>
                <configuration>
                    <skip>true</skip>
                </configuration>
            </plugin>

        </plugins>
    </build>
</project>

依賴這里，方便起見，直接添加flume和spark的全部依賴，自己到maven的官方庫上搜索，然后添加就可以。接著最重要的是 spark使用flume的依賴的spark-streaming-flume 這個包，不要漏了。如果在集群中運(yùn)行，記得將這個包放到spark的jars目錄下

代碼：

package SparkStreamExer

import org.apache.flume.source.avro.AvroFlumeEvent
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

object StreamingFromFlume {
  def main(args: Array[String]): Unit = {
    //設(shè)置日志級別為ERROR，默認(rèn)是INFO
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    val conf = new SparkConf().setAppName("spark streaming from flume").setMaster("local[2]")
    conf.registerKryoClasses(Array())

    val ssc = new StreamingContext(conf, Seconds(4))

    //創(chuàng)建flumeevent，接收從flume push來的數(shù)據(jù)
    val flumeDStream = FlumeUtils.createStream(ssc, "192.168.50.1", 1234, StorageLevel.MEMORY_ONLY)

    val eventDStream = flumeDStream.map(event => {
      (event.event.getHeaders.toString,new String(event.event.getBody.array()))
    })

    eventDStream.print()

    ssc.start()
    ssc.awaitTermination()
  }

}

（3）啟動：

先啟動spark程序，直接在ide中運(yùn)行。
接著啟動flume：flume-ng agent --conf conf --name a1 --conf-file conf/flume-spark.properties  -Dflume.root.logger=INFO,console

然后自己在監(jiān)控目錄下修改文件，或者添加文件。
接著查看ide中輸出的數(shù)據(jù)

2、spark從flume拉取數(shù)據(jù)
這種方式比起第一種方式要更加靈活，可擴(kuò)展性高。
（1）flume配置文件

a1.sources=r1
a1.channels=c1
a1.sinks=k1

a1.sources.r1.type=TAILDIR
a1.sources.r1.filegroups=f1
a1.sources.r1.filegroups.f1=/opt/modules/apache-flume-1.8.0-bin/logs/.*
a1.sources.r1.fileHeader=true

a1.channels.c1.type=memory
a1.channels.c1.capacity=10000
a1.channels.c1.transactionCapacity=100

# 這里使用spark自己實(shí)現(xiàn)的一個sink
a1.sinks.k1.type=org.apache.spark.streaming.flume.sink.SparkSink
a1.sinks.k1.hostname=192.168.50.121
a1.sinks.k1.port=1234

a1.sources.r1.channels=c1
a1.sinks.k1.channel=c1

另外，需要將spark-streaming-flume-sink_2.11-2.1.0.jar 這個jar包添加到flume的lib目錄下，這是上面使用的SparkSink所在的jar包?？梢宰约涸趇dea中添加這個依賴，然后下載，接著到本地倉庫目錄復(fù)制到flume的lib下。
（2）代碼
pom.xml

和上面類似，只是多了
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-flume-sink_2.11</artifactId>
<version>2.1.0</version>
</dependency>
這個依賴

代碼：

package SparkStreamExer

import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.flume.FlumeUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

object FromFlumePull {
  def main(args: Array[String]): Unit = {
    //設(shè)置日志級別為ERROR，默認(rèn)是INFO
    Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)

    val conf = new SparkConf().setAppName("flume through spark sink").setMaster("local[2]")

    val ssc = new StreamingContext(conf, Seconds(4))

    //創(chuàng)建 poll streaming，從flume拉取數(shù)據(jù)到本地處理
    val flumePollingStream = FlumeUtils.createPollingStream(ssc, "bigdata121", 1234, StorageLevel.MEMORY_ONLY)

    /**
      * 這里要注意：
      * event.event.getBody.array() 不要直接 toString，解析處理的字符串只是[class name]@[hashCode]的形式
      * 應(yīng)該用 New string(event.event.getBody.array()) 這樣會根據(jù)默認(rèn)編解碼規(guī)則給bytes字符串解碼
      * 因?yàn)閭鬏斶^來的是bytes數(shù)據(jù)
      */
    flumePollingStream.map(event=>{
      (event.event.getHeaders.toString, new String(event.event.getBody.array()))
    }).print()

    ssc.start()
    ssc.awaitTermination()
  }
}

（3）啟動
啟動方式和上面類似，這里不重復(fù)。

（4）遇到的問題
問題1：
已經(jīng)將spark-streaming-flume-sink_2.11.jar包放到flume的lib目錄下，flume的agent啟動時報(bào)錯：

29 Aug 2019 17:59:31,838 WARN  [Spark Sink Processor Thread - 10] (org.apache.spark.streaming.flume.sink.Logging$class.logWarning:80)  - Error while processing transaction.
java.lang.IllegalStateException: begin() called when transaction is OPEN!
        at com.google.common.base.Preconditions.checkState(Preconditions.java:145)
        at org.apache.flume.channel.BasicTransactionSemantics.begin(BasicTransactionSemantics.java:131)
        at org.apache.spark.streaming.flume.sink.TransactionProcessor$$anonfun$populateEvents$1.apply(TransactionProcessor.scala:114)
        at org.apache.spark.streaming.flume.sink.TransactionProcessor$$anonfun$populateEvents$1.apply(TransactionProcessor.scala:113)
        at scala.Option.foreach(Option.scala:236)
        at org.apache.spark.streaming.flume.sink.TransactionProcessor.populateEvents(TransactionProcessor.scala:113)
        at org.apache.spark.streaming.flume.sink.TransactionProcessor.call(TransactionProcessor.scala:243)
        at org.apache.spark.streaming.flume.sink.TransactionProcessor.call(TransactionProcessor.scala:43)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

 重點(diǎn)在：java.lang.IllegalStateException: begin() called when transaction is OPEN!
有可能是flume的一些jar包的問題，具體還不清楚。

屢次報(bào)這個錯，最后看了看flume的lib下的scala包

scala-library-2.10.5.jar

是這個版本，我放進(jìn)去的sparksink包是基于 scala 2.11.8的，所以我在想是不是scala library包版本不對，所以從spark的jar目錄下拷貝scala-library-2.11.8.jar 這個包過去flume下，將原來的重命名，不讓flume使用舊的。
接著重新啟動flume agent，正常運(yùn)行。

所以這個問題是因?yàn)橐蕾嚢姹静粚?yīng)的問題發(fā)生的。

問題2：
讀取body的時候，直接toString和new String有區(qū)別，前者亂碼，后者還原原本字符串

toString()與new String ()用法區(qū)別

str.toString是調(diào)用了b這個object對象的類的toString方法。一般是返回這么一個String：[class name]@[hashCode]。
new String(str)是根據(jù)parameter是一個字節(jié)數(shù)組，使用Java虛擬機(jī)默認(rèn)的編碼格式，將這個字節(jié)數(shù)組decode為對應(yīng)的字符。若虛擬機(jī)默認(rèn)的編碼格式是ISO-8859-1，按照ascii編碼表即可得到字節(jié)對應(yīng)的字符。

什么時候用什么方法呢？
new String（）一般使用字符轉(zhuǎn)碼的時候,byte[]數(shù)組的時候
toString（）將對象打印的時候使用

分享題目：五、spark--sparkstreaming原理和使用
轉(zhuǎn)載注明：http://bm7419.com/article32/ipoipc.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供商城網(wǎng)站、虛擬主機(jī)、手機(jī)網(wǎng)站建設(shè)、、Google、企業(yè)建站

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源：創(chuàng)新互聯(lián)

猜你還喜歡下面的內(nèi)容