SparkStreaming中使用DataFrame和SQL操作

拼搏现实的明天。 2023-02-25 15:23 7阅读 0赞

## 一.简介 ##

可以轻松地对流数据使用DataFrames和SQL操作。StreamingContext和其使用的SparkContext必须来自同一个SparkSession。以便可以在驱动程序故障时重新启动它。这是通过创建SparkSession的延迟实例化单例实例来完成的。在下面的示例中，修改了前面的单词计数示例，以使用DataFrames和SQL生成单词计数。每个RDD都转换为一个DataFrame，注册为临时表，然后使用SQL查询。

## 二.代码实战 ##

package spark2.streaming
    
    import org.apache.log4j.{ Level, Logger}
    import org.apache.spark.sql.SparkSession
    import org.apache.spark.streaming.{ Seconds, StreamingContext}
    
    /** * 实时统计 * Created by Administrator on 2020/07/13. */
    object WordCountByForeachRDD { 
      Logger.getLogger("org").setLevel(Level.WARN)
      def main(args: Array[String]) { 
        val spark = SparkSession.builder()
          .appName(s"${this.getClass.getSimpleName}")
          .master("local[2]")
          .getOrCreate()
        val sc = spark.sparkContext
        /** * 5s计算一次 */
        val ssc = new StreamingContext(sc, Seconds(5))
    
        /** * 加载socket数据 */
        val socketDStream = ssc.socketTextStream("master", 9999)
        
        val words = socketDStream.flatMap(_.split(" "))
    
        words.foreachRDD(rdd =>{ 
          val len = rdd.count()
          if(len > 0){ 
            val spark = SparkSession.builder().config(rdd.sparkContext.getConf).getOrCreate()
            import spark.implicits._
    
            /** * 转换为df，并创建临时视图 */
            val wordDf = rdd.toDF("word")
            wordDf.createOrReplaceTempView("words")
    
            /** * 执行SQL语句 */
            val result = spark.sql("select word,count(*) as total from words group by word")
            result.show()
          }
        })
    
        /** * 开启流计算 */
        ssc.start()
        ssc.awaitTermination()
      }
    }

执行结果：  
![在这里插入图片描述][20200713154204802.png]

## 三.必备知识 ##

还可以在来自不同线程的流数据定义的表上运行SQL查询【即与正在运行的StreamingContext异步执行】。只需确保将StreamingContext设置为记住足够的流数据即可运行查询。否则，不知道任何异步SQL查询的StreamingContext将在查询完成之前删除旧的流数据。例如，如果要查询最后一批，但是查询可能需要5分钟才能运行完成，此时必须要调用streamingContext.remember(Minutes(5))，来保存5分钟内的流数据，确保数据不丢失。

[20200713154204802.png]: /images/20230209/d5dc7e125a4a48adbd5ba0fd43eec5ff.png