sparkstreaming和sparksql的整合

使用idea编程

新建一个maven项目，添加scala语言环境，并且在resources中添加log4J.properties

pom.xml依赖是

 
 
     4.0.0
 ​
     org.example
     sz2103-sparkstreaming
     1.0-SNAPSHOT
     
         
             org.apache.spark
             spark-streaming_2.11
             2.2.3
         
         
             org.apache.spark
             spark-streaming-kafka-0-10_2.11
             2.2.3
         
         
             redis.clients
             jedis
             3.0.0
         
         
             org.apache.spark
             spark-sql_2.11
             2.2.3
         
     
 ​
 

 package com.qf.day03
 ​
 import java.util.logging.{Level, Logger}
 ​
 import org.apache.spark.SparkConf
 import org.apache.spark.streaming.dstream.ReceiverInputDStream
 import org.apache.spark.streaming.{Durations, StreamingContext}
 ​
 ​
 object sparkstreamigAndSsql {
   def main(args: Array[String]): Unit = {
     val conf = new SparkConf().setMaster("local[*]").setAppName("ss")
     val ssc = new StreamingContext(conf,Durations.seconds(10))
      val dStream:ReceiverInputDStream[String] = ssc.socketTextStream("qianfeng01",10086)    //主机号  端口
     
     
     dStream.print()
     ssc.start()
     ssc.awaitTermination()
 ​
   }
 }
 ​

在idea运行这个代码之前，要开启主机号对应的虚拟机，然后敲下 nc -l 10086

 package com.qf.day03
 ​
 import java.util.logging.{Level, Logger}
 ​
 import org.apache.spark.SparkConf
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.{Dataframe, SparkSession}
 import org.apache.spark.streaming.dstream.ReceiverInputDStream
 import org.apache.spark.streaming.{Durations, StreamingContext}
 ​
 object sparkstreamigAndSsql {
   def main(args: Array[String]): Unit = {
     val conf = new SparkConf().setMaster("local[*]").setAppName("ss")
     val ssc = new StreamingContext(conf,Durations.seconds(10))
     val sparkSession:SparkSession = SparkSession.builder().config(conf).getOrCreate()
     import sparkSession.implicits._
      val dStream:ReceiverInputDStream[String] = ssc.socketTextStream("qianfeng01",10086)
     dStream.window(Durations.minutes(1),Durations.seconds(20)).foreachRDD(rdd=>{
       //将rdd转成四列形式的rdd
      val rdd1: RDD[(String,String,String,Int)] = rdd.map(line=>{
        val arr:Array[String] =line.split(" ")
        (arr(0),arr(1),arr(2),arr(3).toInt)
      })
       //rdd-》 DF
       val df:Dataframe = rdd1.toDF("time","product_id","product_name","num")
       //构建表
       df.createOrReplaceTempView("product_sale_info")
       val sql =
         """
           |select *
           |from
           |   (select t1.*,dense_rank() over(order by total desc) rk
           |   from
           |     (
           |     select product_id,product_name,sum(num) total
           |     from product_sale_info
           |      group by product_id,product_name) t1
           |   ) t2
           |  where rk < 4
           |""".stripMargin
       sparkSession.sql(sql).show()
     })
 ​
     ssc.start()
     ssc.awaitTermination()
 ​
   }
 }
 ​

在执行linux运行nc -l 10086,

运行上面编写的api

然后在nc指令后面继续敲

 8:00 1001 毛衣 10
 8:00 1002 毛衣 1
 8:00 1003 毛衣 10
 8:00 1004 毛衣 10
 8:00 1005 毛衣 10