Spark Streaming流式處理

Spark Streaming介紹

Spark Streaming概述

• Spark Streaming makes it easy to build scalable fault-tolerant streaming applications.

• 它可以非常容易的構建一個可擴展、具有容錯機制的流式應用。

• 對接很多的外部數據源

  • Kafka、Flume、Twitter、ZeroMQ和簡單的TCP套接字（socket）等等

 

Spark Streaming特性

• 1、易用性

  • 可以像編寫離線批處理一樣去編寫流式程序，支持java/scala/python語言

• 2、容錯性

  • SparkStreaming在沒有額外代碼和配置的情況下可以恢復丟失的工作。

• 3、可以融合到spark體系

• 流式處理與批處理和交互式查詢相結合。

 

Spark Streaming原理介紹

• 1、sparkStreaming原理

  • Spark Streaming 是基於spark的流式批處理引擎，其基本原理是把輸入數據以某一時間間隔批量的處理，當批處理間隔縮短到秒級時，便可以用於處理實時數據流。

• 2、Spark Streaming計算流程

  • Dstream是sparkStreaming底層抽象，它封裝了一個個批次的數據，針對於Dstream做相應操作，這個時候這些方法就作用在Dstream中RDD

• 3、SparkStreaming容錯性

  • 依賴RDD的血統+數據源端的安全性

• 4、SparkStreaming實時性

  • SparkStreaming實時性不是特別高，它是以某一時間批次進行處理，批次最小時間0.5s - 2s

  • 其最小的Batch Size的選取在0.5~2秒鍾之間（Storm目前最小的延遲是100ms左右），所以Spark Streaming能夠滿足除對實時性要求非常高（如高頻實時交易）之外的所有流式准實時計算場景。

 

Dstream

　　Discretized Stream是Spark Streaming的基礎抽象，代表持續性的數據流和經過各種Spark算子操作后的結果數據流。在內部實現上，DStream是一系列連續的RDD來表示。每個RDD含有一段時間間隔內的數據

DStream相關操作

 　　DStream上的操作與RDD的類似，分為Transformations（轉換）和Output Operations（輸出）兩種，此外轉換操作中還有一些比較特殊的操作，如：updateStateByKey()、transform()以及各種Window相關的操作

 transformation

它是一個轉換，它把一個Dstream轉換生成一個Dstream，它也不會觸發任務真正運行

 

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

 

 

特殊的Transformations

（1）UpdateStateByKey Operation

　　UpdateStateByKey用於記錄歷史記錄，保存上次的狀態

（2）Window Operations(開窗函數)

滑動窗口轉換操作:

　　滑動窗口轉換操作的計算過程如下圖所示，我們可以事先設定一個滑動窗口的長度（也就是窗口的持續時間），並且設定滑動窗口的時間間隔（每隔多長時間執行一次計算），然后，就可以讓窗口按照指定時間間隔在源DStream上滑動，每次窗口停放的位置上，都會有一部分DStream被框入窗口內，形成一個小段的DStream，這時，就可以啟動對這個小段DStream的計算。

 

Output Operations

Output Operations可以將DStream的數據輸出到外部的數據庫或文件系統，當某個Output Operations被調用時（與RDD的Action相同），spark streaming程序才會開始真正的計算過程。

 

Output Operation	Meaning
print()	打印到控制台
saveAsTextFiles(prefix, [suffix])	保存流的內容為文本文件，文件名為 "prefix-TIME_IN_MS[.suffix]".
saveAsObjectFiles(prefix, [suffix])	保存流的內容為SequenceFile，文件名為  "prefix-TIME_IN_MS[.suffix]".
saveAsHadoopFiles(prefix, [suffix])	保存流的內容為hadoop文件，文件名為  "prefix-TIME_IN_MS[.suffix]".
foreachRDD(func)	對Dstream里面的每個RDD執行func

 

 

 

DStream操作實戰

一、

 SparkStreaming接受socket數據，實現單詞計數WordCount

 

（1）安裝並啟動生產者

首先在linux服務器上用YUM安裝nc工具，nc命令是netcat命令的簡稱,它是用來設置路由器。我們可以利用它向某個端口發送數據。

　　yum install -y nc

（2）通過netcat工具向指定的端口發送數據

      nc -lk 9999  

（3）編寫Spark Streaming程序

 導入依賴

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

代碼開發

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}

//todo:需求：利用sparkStreaming接受socket數據，實現單詞統計WordCount
object SparkStreamingSocket {
  def main(args: Array[String]): Unit = {
      //1、創建SparkConf
      val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingSocket").setMaster("local[2]")
      //2、創建SparkContext
      val sc = new SparkContext(sparkConf)
      sc.setLogLevel("WARN")
      //3、創建StreamingContext對象，需要2個參數，第一是SparkContext，第二個表示批處理時間間隔
      val ssc = new StreamingContext(sc,Seconds(5))
      //4、通過StreamingContext對象來接受socket數據
      val socketTextStream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.200.100",9999)
      //5、切分每一行
      val wordsDstream: DStream[String] = socketTextStream.flatMap(_.split(" "))
      //6、每一個單詞計為1
      val wordAndOneDstream: DStream[(String, Int)] = wordsDstream.map((_,1))
     //7、相同單詞出現的次數累加
      val result: DStream[(String, Int)] = wordAndOneDstream.reduceByKey(_+_)
     //8、打印結果
     result.print()

    //9、開啟流式計算
      ssc.start()
      //該方法為一直阻塞在這里，等待程序結束
      ssc.awaitTermination()
  }
}

運行程序

 

 在liunx中輸入數據

 

 

 

二、

SparkStreaming接受socket數據，實現所有批次單詞計數結果累加

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

//todo:需求：利用sparkStreaming接受socket數據，實現所有批次單詞統計結果累加
object SparkStreamingSocketTotal {
  /**
    *  (hadoop,1) (hadoop,1) (hadoop,1) (hadoop,1)
    * @param currentValues  它表示在當前批次中相同單詞出現所有的1,把這所有的1封裝在一個list集合中 （hadoop,List(1,1,1,1)）
    * @param historyValues  它表示在之前所有批次中每一個單詞出現的次數
    * @return
    */
  def updateFunc(currentValues:Seq[Int], historyValues:Option[Int]):Option[Int] = {
                val newValue: Int = currentValues.sum + historyValues.getOrElse(0)
                Some(newValue)
  }

  def main(args: Array[String]): Unit = {
      //1、創建SparkConf
      val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingSocketTotal").setMaster("local[2]")
       //2、創建SparkContext
      val sc = new SparkContext(sparkConf)
      sc.setLogLevel("WARN")
      //3、創建StreamingContext對象
      val ssc = new StreamingContext(sc,Seconds(5))

       //需要設置一個checkpoint，由於保存每一個批次中間結果數據
       ssc.checkpoint("./socket")

      //4、接受socket是數據
      val socketTextStream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.200.100",9999)
      //5、切分每一行
      val words: DStream[String] = socketTextStream.flatMap(_.split(" "))
      //6、每個單詞計為1
      val wordAndOne: DStream[(String, Int)] = words.map((_,1))
      //7、相同單詞所有批次結果累加
      val result: DStream[(String, Int)] = wordAndOne.updateStateByKey(updateFunc)

    //8、打印結果
    result.print()
    //9、開啟流式計算
    ssc.start()
    //該方法為一直阻塞在這里，等待程序結束
    ssc.awaitTermination()
  }
}

 

三、

SparkStreaming開窗函數reduceByKeyAndWindow，實現單詞計數

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

//todo:需求：利用sparkStreaming接受socket數據，使用reduceByKeyAndWindow開窗函數實現單詞統計
object SparkStreamingSocketWindow {
  def main(args: Array[String]): Unit = {
    //1、創建SparkConf
    val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingSocketWindow").setMaster("local[2]")
    //2、創建SparkContext
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")
    //3、創建StreamingContext對象，需要2個參數，第一是SparkContext，第二個表示批處理時間間隔
    val ssc = new StreamingContext(sc,Seconds(5))
    //4、通過StreamingContext對象來接受socket數據
    val socketTextStream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.200.100",9999)
    //5、切分每一行
    val wordsDstream: DStream[String] = socketTextStream.flatMap(_.split(" "))
    //6、每一個單詞計為1
    val wordAndOneDstream: DStream[(String, Int)] = wordsDstream.map((_,1))
    //7、相同單詞出現的次數累加
      //方法中需要三個參數
       //reduceFunc：第一個就是一個函數
       //windowDuration：第二個表示窗口長度
       //slideDuration :第三個表示滑動窗口的時間間隔，也就意味着每隔多久計算一次
    val result: DStream[(String, Int)] = wordAndOneDstream.reduceByKeyAndWindow((x:Int,y:Int)=>x+y,Seconds(10),Seconds(5))

    //8、打印結果
    result.print()

    //9、開啟流式計算
    ssc.start()
    //該方法為一直阻塞在這里，等待程序結束
    ssc.awaitTermination()
  }
}

 

四、

SparkStreaming開窗函數統計一定時間內的熱門詞匯

transform

• 實現一個Dstream轉換成一個新的Dstream，當前這個方法需要一個函數，這個函數的輸入參數是前面Dstream中的rdd,函數的返回值是一個新的RDD

import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

//todo:需求：利用sparkStreaming接受socket數據，通過reduceByKeyAndWindow實現一定時間內熱門詞匯
object SparkStreamingSocketWindowHotWords {
  def main(args: Array[String]): Unit = {
       //1、創建SparkConf
      val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingSocketWindowHotWords").setMaster("local[2]")
      //2、創建SparkContext
        val sc = new SparkContext(sparkConf)
        sc.setLogLevel("WARN")
     //3、創建StreamingContext
      val ssc = new StreamingContext(sc,Seconds(5))
    //4、接受socket數據
      val socketTextStream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.200.100",9999)
    //5、切分每一行獲取所有的單詞
      val words: DStream[String] = socketTextStream.flatMap(_.split(" "))
    //6、每個單詞計為1
        val wordAndOne: DStream[(String, Int)] = words.map((_,1))
    //7、相同單詞出現次數累加
      val result: DStream[(String, Int)] = wordAndOne.reduceByKeyAndWindow((x:Int,y:Int)=>x+y,Seconds(10),Seconds(10))
    //8、按照單詞出現的次數降序排序
    val sortedDstream: DStream[(String, Int)] = result.transform(rdd => {
      //按照單詞次數降序
      val sortedRDD: RDD[(String, Int)] = rdd.sortBy(_._2, false)
      //取出出現次數最多的前3位
      val top3: Array[(String, Int)] = sortedRDD.take(3)
      //打印
      println("=================top3===============start")
      top3.foreach(println)
      println("=================top3===============end")
      sortedRDD
    })

    sortedDstream.print()
    //開啟流式計算
      ssc.start()
      ssc.awaitTermination()
  }
}

 

Spark Streaming整合flume

　　flume作為日志實時采集的框架，可以與SparkStreaming實時處理框架進行對接，flume實時產生數據，sparkStreaming做實時處理。

　　Spark Streaming對接FlumeNG有兩種方式，一種是FlumeNG將消息Push推給Spark Streaming，還有一種是Spark Streaming從flume 中Poll拉取數據。

Poll方式

（1）安裝flume1.6以上

（2）下載依賴包

spark-streaming-flume-sink_2.11-2.0.2.jar放入到flume的lib目錄下

（3）修改flume/lib下的scala依賴包版本

從spark安裝目錄的jars文件夾下找到scala-library-2.11.8.jar 包，替換掉flume的lib目錄下自帶的scala-library-2.10.1.jar。

（4）寫flume的agent，注意既然是拉取的方式，那么flume向自己所在的機器上產數據就行

（5）編寫flume-poll-spark.conf配置文件

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

#source
a1.sources.r1.channels = c1
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /root/data
a1.sources.r1.fileHeader = true

#channel
a1.channels.c1.type =memory
a1.channels.c1.capacity = 20000
a1.channels.c1.transactionCapacity=5000

#sinks
a1.sinks.k1.channel = c1
a1.sinks.k1.type = org.apache.spark.streaming.flume.sink.SparkSink
a1.sinks.k1.hostname=node1
a1.sinks.k1.port = 8888
a1.sinks.k1.batchSize= 2000

啟動flume

bin/flume-ng agent -n a1 -c conf -f conf/flume-poll-spark.conf -Dflume.root.logger=info,console

代碼實現：

引入依賴

 <!--sparkStreaming整合flume-->
 <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-flume_2.11</artifactId>
            <version>2.0.2</version>
 </dependency>

代碼

import java.net.InetSocketAddress

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.flume.{FlumeUtils, SparkFlumeEvent}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

//todo:需求：利用sparkStreaming整合flume---（poll拉模式）
 //通過poll拉模式整合，啟動順序---》啟動flume----》sparkStreaming程序
object SparkStreamingFlumePoll {
  def main(args: Array[String]): Unit = {
      //1、創建SparkConf
        val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingFlumePoll").setMaster("local[2]")
      //2、創建SparkContext
      val sc = new SparkContext(sparkConf)
      sc.setLogLevel("WARN")
      //3、創建StreamingContext
      val ssc = new StreamingContext(sc,Seconds(5))
      //4、讀取flume中的數據
      val pollingStream: ReceiverInputDStream[SparkFlumeEvent] = FlumeUtils.createPollingStream(ssc,"192.168.200.100",8888)

        //定義一個List集合可以封裝不同的flume收集數據
        val address=List(new InetSocketAddress("node1",8888),new InetSocketAddress("node2",8888),new InetSocketAddress("node3",8888))
        //val pollingStream: ReceiverInputDStream[SparkFlumeEvent] = FlumeUtils.createPollingStream(ssc,address,StorageLevel.MEMORY_AND_DISK_SER_2)

      //5、 event是flume中傳輸數據的最小單元，event中數據結構：{"headers":"xxxxx","body":"xxxxxxx"}
     val flume_data: DStream[String] = pollingStream.map(x => new String(x.event.getBody.array()))
      //6、切分每一行
      val words: DStream[String] = flume_data.flatMap(_.split(" "))
      //7、每個單詞計為1
      val wordAndOne: DStream[(String, Int)] = words.map((_,1))
      //8、相同單詞出現次數累加
      val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)
      //9、打印
      result.print()
      ssc.start()
      ssc.awaitTermination()
  }
}

 

Push方式

編寫flume-push-spark.conf配置文件

#push mode
a1.sources = r1
a1.sinks = k1
a1.channels = c1
#source
a1.sources.r1.channels = c1
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /root/data
a1.sources.r1.fileHeader = true
#channel
a1.channels.c1.type =memory
a1.channels.c1.capacity = 20000
a1.channels.c1.transactionCapacity=5000
#sinks
a1.sinks.k1.channel = c1
a1.sinks.k1.type = avro
a1.sinks.k1.hostname=192.168.1.120
a1.sinks.k1.port = 8888
a1.sinks.k1.batchSize= 2000
[root@node1 conf]# 
[root@node1 conf]# 
[root@node1 conf]# vi flume-push-spark.conf 
#push mode
a1.sources = r1
a1.sinks = k1
a1.channels = c1
#source
a1.sources.r1.channels = c1
a1.sources.r1.type = spooldir
a1.sources.r1.spoolDir = /root/data
a1.sources.r1.fileHeader = true
#channel
a1.channels.c1.type =memory
a1.channels.c1.capacity = 20000
a1.channels.c1.transactionCapacity=5000
#sinks
a1.sinks.k1.channel = c1
a1.sinks.k1.type = avro
a1.sinks.k1.hostname=192.168.75.57
a1.sinks.k1.port = 8888
a1.sinks.k1.batchSize= 2000

注意配置文件中指明的hostname和port是spark應用程序所在服務器的ip地址和端口。

啟動flume

bin/flume-ng agent -n a1 -c conf -f conf/flume-push-spark.conf -Dflume.root.logger=info,console

代碼開發

import java.net.InetSocketAddress

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.flume.{FlumeUtils, SparkFlumeEvent}

//todo:需求：利用sparkStreaming整合flume-------（push推模式）
//push推模式啟動順序：先啟動sparkStreaming程序，然后啟動flume
object SparkStreamingFlumePush {
  def main(args: Array[String]): Unit = {
    //1、創建SparkConf
    val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingFlumePush").setMaster("local[2]")
    //2、創建SparkContext
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")
    //3、創建StreamingContext
    val ssc = new StreamingContext(sc,Seconds(5))
    //4、讀取flume中的數據
    val pollingStream: ReceiverInputDStream[SparkFlumeEvent] = FlumeUtils.createStream(ssc,"192.168.75.57",8888)

    //5、 event是flume中傳輸數據的最小單元，event中數據結構：{"headers":"xxxxx","body":"xxxxxxx"}
    val flume_data: DStream[String] = pollingStream.map(x => new String(x.event.getBody.array()))
    //6、切分每一行
    val words: DStream[String] = flume_data.flatMap(_.split(" "))
    //7、每個單詞計為1
    val wordAndOne: DStream[(String, Int)] = words.map((_,1))
    //8、相同單詞出現次數累加
    val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)
    //9、打印
    result.print()

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

 

Spark Streaming整合kafka

KafkaUtils.createDstream

• 當前是利用kafka高層次api（偏移量由zk維護）

• 這種方式默認數據會丟失，可以通過啟用WAL預寫日志，將接受到的數據同時也寫入了到HDFS中，可以保證數據源端的安全性，當前Dstream中某個rdd的分區數據丟失了，可以通過血統，拿到原始數據重新計算恢復得到。

• 但是它保證不了數據只被處理一次。

引入依賴

<!--sparkStreaming整合kafka-->
<dependency>
       <groupId>org.apache.spark</groupId>
       <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
       <version>2.0.2</version>
</dependency>

代碼開發

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.kafka.KafkaUtils

import scala.collection.immutable

//todo:需求：利用sparkStreaming整合kafka---利用kafka高層次api（偏移量由zk維護）
object SparkStreamingKafkaReceiver {
  def main(args: Array[String]): Unit = {
      //1、創建SparkConf
      val sparkConf: SparkConf = new SparkConf()
                                .setAppName("SparkStreamingKafkaReceiver")
                                .setMaster("local[4]")                                 
　　　　　　　　　　　　　　　　　　　 .set("spark.streaming.receiver.writeAheadLog.enable","true") //開啟WAL日志，保證數據源端的安全性
      //2、創建SparkContext
      val sc = new SparkContext(sparkConf)
      sc.setLogLevel("WARN")
     //3、創建StreamingContext
      val ssc = new StreamingContext(sc,Seconds(5))
       ssc.checkpoint("./kafka-receiver")

     //4、接受topic的數據
        //zk服務地址
       val zkQuorum="node1:2181,node2:2181,node3:2181"
       //消費者組id
      val groupId="sparkStreaming_group"
       //topic信息     //map中的key表示topic名稱，map中的value表示當前針對於每一個receiver接受器采用多少個線程去消費數據
       val topics=Map("itcast" -> 1)
       //(String, String):第一個String表示消息的key，第二個String就是消息具體內容
       // val kafkaDstream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc,zkQuorum,groupId,topics)

      //這里構建了多個receiver接受數據
    val receiverListDstream: immutable.IndexedSeq[ReceiverInputDStream[(String, String)]] = (1 to 3).map(x => {
      val kafkaDstream: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc, zkQuorum, groupId, topics)
      kafkaDstream
    })
        //通過調用streamingContext的union將所有的receiver接收器中的數據合並
      val kafkaDstream: DStream[(String, String)] = ssc.union(receiverListDstream)
    //    kafkaDstream.foreachRDD(rdd =>{
//      rdd.foreach(x=>println("key:"+x._1+" value:"+x._2))
//    })
      //5、獲取topic中的數據
      val kafkaData: DStream[String] = kafkaDstream.map(_._2)

        //6、切分每一行
        val words: DStream[String] = kafkaData.flatMap(_.split(" "))
        //7、每個單詞計為1
        val wordAndOne: DStream[(String, Int)] = words.map((_,1))
        //8、相同單詞出現次數累加
        val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)
        //9、打印
        result.print()

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

 

（1）啟動zookeeper集群

zkServer.sh start

（2）啟動kafka集群

kafka-server-start.sh  /export/servers/kafka/config/server.properties

（3） 創建topic

kafka-topics.sh --create --zookeeper hdp-node-01:2181 --replication-factor 1 --partitions 3 --topic itcast

（4） 向topic中生產數據

通過shell命令向topic發送消息

kafka-console-producer.sh --broker-list hdp-node-01:9092 --topic  itcast

（5）運行代碼,查看控制台結果數據

總結:

　　通過這種方式實現，剛開始的時候系統正常運行，沒有發現問題，但是如果系統異常重新啟動sparkstreaming程序后，發現程序會重復處理已經處理過的數據，這種基於receiver的方式，是使用Kafka的高級API，topic的offset偏移量在ZooKeeper中。這是消費Kafka數據的傳統方式。這種方式配合着WAL機制可以保證數據零丟失的高可靠性，但是卻無法保證數據只被處理一次，可能會處理兩次。因為Spark和ZooKeeper之間可能是不同步的。官方現在也已經不推薦這種整合方式，我們使用官網推薦的第二種方式kafkaUtils的createDirectStream()方式。

 

KafkaUtils.createDirectStream

　　這種方式不同於Receiver接收數據，它定期地從kafka的topic下對應的partition中查詢最新的偏移量，再根據偏移量范圍在每個batch里面處理數據，Spark通過調用kafka簡單的消費者Api（低級api）讀取一定范圍的數據。

相比基於Receiver方式有幾個優點： 
A、簡化並行

不需要創建多個kafka輸入流，然后union它們，sparkStreaming將會創建和kafka分區數相同的rdd的分區數，而且會從kafka中並行讀取數據，spark中RDD的分區數和kafka中的topic分區數是一一對應的關系。

 

B、高效， 

第一種實現數據的零丟失是將數據預先保存在WAL中，會復制一遍數據，會導致數據被拷貝兩次，第一次是接受kafka中topic的數據，另一次是寫到WAL中。而沒有receiver的這種方式消除了這個問題。 

C、恰好一次語義(Exactly-once-semantics)

Receiver讀取kafka數據是通過kafka高層次api把偏移量寫入zookeeper中，雖然這種方法可以通過數據保存在WAL中保證數據不丟失，但是可能會因為sparkStreaming和ZK中保存的偏移量不一致而導致數據被消費了多次。EOS通過實現kafka低層次api，偏移量僅僅被ssc保存在checkpoint中，消除了zk和ssc偏移量不一致的問題。缺點是無法使用基於zookeeper的kafka監控工具。

 

代碼

import kafka.serializer.StringDecoder
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils

//todo:需求：利用sparkStreaming整合kafka，利用kafka低級api（偏移量不在由zk維護）
object SparkStreamingKafkaDirect {
  def main(args: Array[String]): Unit = {
    //1、創建SparkConf
    val sparkConf: SparkConf = new SparkConf()
      .setAppName("SparkStreamingKafkaDirect")
      .setMaster("local[4]")
    //2、創建SparkContext
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")
    //3、創建StreamingContext
    val ssc = new StreamingContext(sc,Seconds(5))
    //此時偏移量由客戶端自己去維護，保存在checkpoint里面
    ssc.checkpoint("./kafka-direct")

    //4、接受topic的數據
        //配置kafka相關參數
      val kafkaParams=Map("bootstrap.servers" ->"node1:9092,node2:9092,node3:9092","group.id" -> "spark_direct","auto.offset.reset" ->"smallest")
        //指定topic的名稱
      val topics=Set("itcast")
    //此時產生的Dstream中rdd的分區跟kafka中的topic分區一一對應
      val kafkaDstream: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics)

    //5、獲取topic中的數據
    val kafkaData: DStream[String] = kafkaDstream.map(_._2)
    //6、切分每一行
    val words: DStream[String] = kafkaData.flatMap(_.split(" "))
    //7、每個單詞計為1
    val wordAndOne: DStream[(String, Int)] = words.map((_,1))
    //8、相同單詞出現次數累加
    val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)
    //9、打印
    result.print()

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

 

StreamingContext.getOrCreate

可以checkpoint目錄來恢復StreamingContext對象

代碼開發

import cn.itcast.streaming.socket.SparkStreamingSocketTotal.updateFunc
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

//todo:利用sparkStreaming接受socket數據，實現整個程序每天24小時運行
//可以允許程序出現異常，再次重新啟動，又可以恢復回來
object SparkStreamingSocketTotalCheckpoint {
  def createStreamingContext(checkpointPath: String): StreamingContext = {
    //1、創建SparkConf
    val sparkConf: SparkConf = new SparkConf().setAppName("SparkStreamingSocketTotalCheckpoint").setMaster("local[2]")
    //2、創建SparkContext
    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")
    //3、創建StreamingContext對象
    val ssc = new StreamingContext(sc,Seconds(5))

    //需要設置一個checkpoint，由於保存每一個批次中間結果數據
    //還會繼續保存這個Driver代碼邏輯，還有任務運行的資源（整個application信息）
    ssc.checkpoint(checkpointPath)
    //4、接受socket是數據
    val socketTextStream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.200.100",9999)
    //5、切分每一行
    val words: DStream[String] = socketTextStream.flatMap(_.split(" "))
    //6、每個單詞計為1
    val wordAndOne: DStream[(String, Int)] = words.map((_,1))
    //7、相同單詞所有批次結果累加
    val result: DStream[(String, Int)] = wordAndOne.updateStateByKey(updateFunc)
    //8、打印結果
    result.print()
    ssc
  }

  def main(args: Array[String]): Unit = {
    val checkpointPath="./ck2018"
    //1、創建StreamingContext
      //通過StreamingContext.getOrCreate方法可以從checkpoint目錄中恢復之前掛掉的StreamingContext
       //第一次啟動程序，最開始這個checkpointPath目錄沒有的數據，就通過后面的函數來幫助我們產生一個StreamingContext
            //並且保存所有數據（application信息）到checkpoint
       //程序掛掉后，第二次啟動程序，它會讀取checkpointPath目錄中數據，就從這個checkpointPath目錄里面來恢復之前掛掉的StreamingContext
       //如果checkpoint目錄中的數據損壞，這個你再次通過讀取checkpoint目錄中的數據來恢復StreamingContext對象不會成功，報異常
    val ssc: StreamingContext = StreamingContext.getOrCreate(checkpointPath, () => {
      val newSSC: StreamingContext = createStreamingContext(checkpointPath)
        newSSC
    })

     //啟動流式計算
      ssc.start()
      ssc.awaitTermination()
  }
}