071 SparkStreaming與SparkSQL集成

1.說明

　　雖然DStream可以轉換成RDD，但是如果比較復雜，可以考慮使用SparkSQL。

 

2.集成方式

　　Streaming和Core整合：
　　　　transform或者foreachRDD方法
　　Core和SQL整合：
　　　　RDD <==> DataFrame 互換

 

3.程序

package com.sql.it
import org.apache.spark.sql.SQLContext
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
object StreamingSQL {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("StreamingWindowOfKafka22")
      .setMaster("local[*]")
    val sc = SparkContext.getOrCreate(conf)
    val ssc = new StreamingContext(sc, Seconds(5))
    // 當調用updateStateByKey函數API的時候，必須給定checkpoint dir
    // 路徑對應的文件夾不能存在
    ssc.checkpoint("hdfs://linux-hadoop01.ibeifeng.com:8020/beifeng/spark/streaming/9421151351")

    val kafkaParams = Map(
      "group.id" -> "streaming-kafka-78912151",
      "zookeeper.connect" -> "linux-hadoop01.ibeifeng.com:2181/kafka",
      "auto.offset.reset" -> "smallest"
    )
    val topics = Map("beifeng" -> 4) // topics中value是讀取數據的線程數量，所以必須大於等於1
    val dstream = KafkaUtils.createStream[String, String, kafka.serializer.StringDecoder, kafka.serializer.StringDecoder](
      ssc, // 給定SparkStreaming上下文
      kafkaParams, // 給定連接kafka的參數信息 ===> 通過Kafka HighLevelConsumerAPI連接
      topics, // 給定讀取對應topic的名稱以及讀取數據的線程數量
      StorageLevel.MEMORY_AND_DISK_2 // 指定數據接收器接收到kafka的數據后保存的存儲級別
    ).map(_._2)

    /**
      * transform：將DStream的操作轉換為RDD的操作，調用該api最終只需要返回一個新的RDD即可
      */
    dstream.transform(rdd => {
      // 使用sql統計wordcoount
      val sqlContext = SQLContextSingelton.getSQLContext(rdd.sparkContext)
      import sqlContext.implicits._
      val procedRDD = rdd.filter(_.nonEmpty).flatMap(_.split(" ").map((_, 1)))
      procedRDD.toDF("word", "c").registerTempTable("tb_word")
      val resultRDD = sqlContext.sql("select word, count(c) as vc from tb_word group by word").map(row => {
        val word = row.getAs[String]("word")
        val count = row.getAs[Long]("vc")
        (word, count)
      })

      resultRDD
    }).print()

    // 啟動開始處理
    ssc.start()
    ssc.awaitTermination() // 等等結束，監控一個線程的中斷操作
  }
}

object SQLContextSingelton {
  @transient private var instance: SQLContext = _

  def getSQLContext(sc: SparkContext): SQLContext = {
    if (instance == null) {
      synchronized[SQLContext] {
        if (instance == null) {
          instance = new SQLContext(sc)
        }
        instance
      }
    }
    instance
  }
}

 

4.效果