streaming通过sql实现wordcount代码

╰+攻爆jí腚メ 2022-03-02 16:48 158阅读 0赞

代码

package com.badou.sql
    
    import org.apache.spark.SparkConf
    import org.apache.spark.SparkContext
    import org.apache.spark.rdd.RDD
    import org.apache.spark.streaming.{Time, Seconds, StreamingContext}
    import org.apache.spark.util.IntParam
    import org.apache.spark.sql.SQLContext
    import org.apache.spark.storage.StorageLevel
    
    case class Record(word: String)
    
    object SQLContextSingleton {
      @transient  private var instance: SQLContext = _
      def getInstance(sparkContext: SparkContext): SQLContext = {
        if (instance == null) {
          instance = new SQLContext(sparkContext)
        }
        instance
      }
    }
    
    object sqlAndStreamingWC {
      def main(args: Array[String]) {
        if (args.length < 2) {
          System.err.println("Usage: NetworkWordCount <hostname> <port>")
          System.exit(1)
        }
    
        val sparkConf = new SparkConf().setMaster("local[2]").setAppName("sqlAndStreamingWC")
        val sc = new SparkContext(sparkConf)
        val ssc = new StreamingContext(sc, Seconds(30))
    
        val lines = ssc.socketTextStream(args(0), args(1).toInt, StorageLevel.MEMORY_AND_DISK_SER)
        val words = lines.flatMap(_.split(" "))
    
        words.foreachRDD((rdd: RDD[String], time: Time) => {
    
          val sqlContext = SQLContextSingleton.getInstance(rdd.sparkContext)
          import sqlContext.implicits._
    
          val wordsDataFrame = rdd.map(w => Record(w)).toDF()
          wordsDataFrame.registerTempTable("words")
          val wordCountsDataFrame =
            sqlContext.sql("select word, count(*) as total from words group by word")
          println(s"========= $time =========")
          wordCountsDataFrame.show()
    
    
        })
    
        ssc.start()
        ssc.awaitTermination()
      }
    }

测试：

![20190320222135199.png][]

结果：

![20190320221803204.png][]

[20190320222135199.png]: /images/20220302/ffe6b2738b7443f981f68d96ea58c376.png
[20190320221803204.png]: /images/20220302/fb1976f3d946403b8b2bb367cfd39561.png