spark sql: 变量共享，文本内容存入数据库

ゝ一纸荒年。 2022-04-22 02:32 200阅读 0赞

现有数据格式如下：（csv格式，10G）

#province-city-reqapp.txt
    广西,桂州,reqmode=false,processmode=true,app3
    广东,深圳,reqmode=true,processmode=false,app1
    湖南,长沙,reqmode=true,processmode=true,app2
    
    #app.txt
    app1,爱奇艺
    app2,搜狐
    app3,新浪微博
    ......

如何使用spark sql统计相关数据： 各省市中各app的reqmode数，process数？

# 1，文件转parquet格式 #

把文件转换为"表", 以便使用sql查询：

object File2parquet_pc_reqcount {
    
      case class province_city_req2(val province: String, val city: String,
                                  val reqmode: String, val procemode: String,val app:String)
    
      def main(args: Array[String]): Unit = {
        //spark会话
        val conf = new SparkConf()
        conf.setAppName(s"${this.getClass.getSimpleName}")
        conf.setMaster("local[*]")
        //序列化器
        conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
        //文件压缩格式
        conf.set("spark.sql.parquet.compression.codec","snappy")
        val spark = SparkSession.builder().config(conf).getOrCreate()
    
        //调用方法
        //spark.sql查询： 数据转换为sql适用文件格式： df--parquet:存储
        val rdd = spark.sparkContext.textFile("Myproject/data/province-city-reqapp.txt")
        //湖南,长沙,reqmode=true,processmode=false  ==>统计各省市的用户分布, 有效请求,原始请求,广告请求
        var rdd2 = rdd.map(line => {
          val array = line.split(",").filter(_.length < 85)
          province_city_req2(array(0), array(1), array(2), array(3),array(4))
        })
    
        //rddd转 DataFrame
        import spark.implicits._
        val df = rdd2.toDF()
        df.repartition(1).write.mode(SaveMode.Append).parquet("Myproject/data/parquet2_2")
    
        //关闭资源
        spark.close()
      }
    }

# 2 ，spark变量共享的方法： #

*  广播变量

val apprdd = spark.sparkContext.textFile("Myproject/data/app.txt")
    val appDict =apprdd.map(x=>{
      val arr = x.split(",")
      (arr(0),arr(1))
    }).collect().toMap
    
    //广播变量
    val appdictBroad = spark.sparkContext.broadcast(appDict)
    
    //在rdd中使用
    rdd.map(x=>{
     val apprealName = appdictBroad.value(appname)
     .....
    })

*  通过redis读写  
    添加maven依赖

<dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>2.9.0</version>
    </dependency>

object Redistool {
      //创建redis连接
      val pool = new JedisPool(new GenericObjectPoolConfig(), "127.0.0.1", 6379)
      def getJedis(): Jedis = {
        val jedis = pool.getResource
        jedis
      }
    }
    
     val apprdd = spark.sparkContext.textFile("Myproject/data/app.txt")
        val appDict =apprdd.map(x=>{
          val arr = x.split(",")
          (arr(0),arr(1))
        }).foreachPartition(it=>{
          it.foreach(y=>{
            val appid=y._1
            val appName=y._2
    
            val jedis = Redistool.getJedis()
            jedis.hset("appdict",appid,appName)
            jedis.close()
          })
        })
        
    //在rdd中使用
    rdd.map(x=>{
      val apprealName = jedis.hget("appdict",appname)
     .....
    })

# 3，使用spark sql: 执行hive查询 #

小文件直接当成meta数据，存入redis, 在join查询时，直接读取：

/**
      * 统计： 各省市的数据分布--》json数据：  本地磁盘
      */
    object CountProvinceCity_rdd_app_redis {
    
      case class province_city_cntrdd(val province: String, val city: String, val appanme:String,
                                      val reqcnt: Int, val procescnt: Int, val adreq:Int)
    
      def main(args: Array[String]): Unit = {
        //读取文件
        //spark会话
        val conf = new SparkConf()
        conf.setAppName(s"${this.getClass.getSimpleName}")
        conf.setMaster("local[*]")
        //序列化器
        conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        //文件压缩格式
        conf.set("spark.sql.parquet.compression.codec", "snappy")
        val spark = SparkSession.builder().config(conf).getOrCreate()
    
        //文件2:
        //app3,新浪微博
       val apprdd = spark.sparkContext.textFile("Myproject/data/app.txt")
       val appDict =apprdd.map(x=>{
        val arr = x.split(",")
          (arr(0),arr(1))
        }).foreachPartition(it=>{
          it.foreach(y=>{
            val appid=y._1
            val appName=y._2
    
            val jedis = Redistool.getJedis()
            jedis.hset("appdict",appid,appName)
            jedis.close()
          })
        })
    
    
        //广播变量
    //    val appdictBroad = spark.sparkContext.broadcast(appDict)
    
        //调用方法
        //rdd查询： 写出mysql
        val rdd1 = spark.sparkContext.textFile("Myproject/data/province-city-reqapp.txt")
        //文件1  |江西,九江,reqmode=true,processmode=true,app3
        //处理： 字段映射--初步过滤处理为1,0
        var rdd2 = rdd1.map(line => {
          val array = line.split(",").filter(_.length < 85)
          val req = array(2)
          val proc = array(3)
    
          //判断是否： 符合要求--true
          var reqcnt = 0
          var procnt = 0
    
          if (req.toLowerCase().contains("true")) {
            reqcnt = 1
          }
          if (proc.toLowerCase().contains("true")) {
            procnt = 1
          }
    
          val appname = array(4)
          val jedis = Redistool.getJedis()
          val apprealName = jedis.hget("appdict",appname)
          jedis.close()
    
          ((array(0), array(1),apprealName), (reqcnt, procnt,  0,0,0))
        }).reduceByKey((x, y) => {
          //取出req, proc数据
          val req_sum = x._1
          val proc_sum = x._2
    
          val req_sum2 = y._1
          val proc_sum2 = y._2
    
          val reqcnt = req_sum + req_sum2
          val procnt = proc_sum + proc_sum2
    
          (req_sum + req_sum2, proc_sum + proc_sum2, 0,0,0)
        })
    
        val rdd3=rdd2.map(x => {
          //内容： [广西,桂州]，  [0,1 , 0,0,0] ==> 广西,桂州, 0,1
          // 处理： 统计： 原始请求数(1,>=1) ， 有效请求数(1,>=2)， 广告请求数(1,3)
          val pro = x._1._1
          val city = x._1._2
          val appname = x._1._3
    
          val reqcnt = x._2._1
          val procnt = x._2._2
    
          var rowreq=0
          var realreq=0
          var adreq=0
    
    
          if (reqcnt == 1 && procnt >= 1) {
            rowreq=1
          }
          if (reqcnt == 1 && procnt >= 2) {
            realreq=1
          }
          if (reqcnt == 1 && procnt ==3) {
            adreq=1
          }
          ((pro, city,appname),(rowreq, realreq,adreq))
        })
    
        //内容：(rowreq, realreq,adreq ）
        //统计映射的数据
       val rdd4= rdd3.reduceByKey((x,y)=>{
          val rowreq = x._1
          val realreq = x._2
          val adreq=x._3
    
          val rowreq2 = y._1
          val realreq2 = y._2
          val adreq2=y._3
    
          (rowreq+rowreq2, realreq+realreq2,adreq+adreq2)
        }).map(x=>{
          province_city_cntrdd(x._1._1, x._1._2,x._1._3,
            x._2._1, x._2._2, x._2._3)
        })
    
        //数据内容：  江西,九江,2,3
        //数据转储： mysql
        val props = new Properties()
        val load = ConfigFactory.load()
        props.setProperty("user", load.getString("jdbc.user"))
        props.setProperty("password", load.getString("jdbc.password"))
    
        //rdd---> dataFrame
        import spark.implicits._
        val df = rdd4.toDF()
        //内容：      province|city|reqcnt|procescnt|
        //                广西|  桂州|     0|       0|
        df.write.mode(SaveMode.Overwrite).
          jdbc(load.getString("jdbc.url"),
            load.getString("jdbc.table_pccount"),
            props)
    
        //关闭资源
        spark.close()
      }
    }