spark同时支持sparksql和sparkStreaming场景

青旅半醒 2023-07-24 02:39 15阅读 0赞

package com.zgw.spark.recommander
    
    import java.util.Properties
    
    import com.zgw.spark.utils.ConfigUtil
    import org.apache.spark.SparkConf
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
    import org.apache.spark.streaming.{Seconds, StreamingContext}
    
    /**
      *
      * 根据用户的视频浏览历史，进行相关的推荐
      * http://lxw1234.com/archives/2015/04/136.htm
      *
      * 实时计算结果
      *
      *
      */
    object RecommendVideoRt {
      def main(args: Array[String]): Unit = {
    
        val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RecommendVideoRt").set("spark.testing.memory", "2147480000")
        //sparkSession,不能直接new，私有的对象,具体看源码
    
        //分析环境对象以及采集周期
        val ssc = new StreamingContext(sparkConf,Seconds(180))
    
    
    //    ssc.addStreamingListener()
    //    ssc.socketStream()
       val aa: RDD[Long] =  ssc.sparkContext.range(0,100,2)
    
        aa.foreach(println)
    
        val spark: SparkSession = SparkSession.builder.config(sparkConf).getOrCreate()
    
    
        val driver = "com.mysql.jdbc.Driver"
        val url = ConfigUtil.getString("jdbc.url")
        val userName =  ConfigUtil.getString("jdbc.username")
        val passWd =  ConfigUtil.getString("jdbc.password")
    
        val readConnProperties4 = new Properties()
        readConnProperties4.put("driver", "com.mysql.jdbc.Driver")
        readConnProperties4.put("user", userName)
        readConnProperties4.put("password", passWd)
    
        val userVideoHistory: DataFrame = spark.read.jdbc(
    //    val userVideoHistory: DataFrame = sqlcontext.read.jdbc(
          url
          ,"(select * from user_video_history ) t"
          ,readConnProperties4
        )
    //    saveAsFileByPartition(dataFrame, "hdfs://gtdata-test01:8020/testw/history_data","|",SaveMode.Overwrite)
    
        userVideoHistory.show(100)
        userVideoHistory.cache()
        userVideoHistory.createTempView("history_data")
     }
    }