sparkSql读取数据库并作相应的条件操作

た入场券 2022-11-26 14:57 78阅读 0赞

# 一、数据库数据 #

# ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h5al9raW5n_size_16_color_FFFFFF_t_70][] #

# 二、代码 #

package com.cn.dataFram
    
    import com.cn.dataFram.DfSaveToMysql.jdbcConnectInfo
    import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
    import org.apache.spark.{SparkConf, SparkContext}
    import org.apache.spark.sql.functions._
    
    object ReadDataToDf {
      case class jdbcConnectInfo(talbeName: String, jdbc: String, user: String, password: String) {
        def get_jdbc = {
          if (jdbc.contains("mysql")) {
            "com.mysql.jdbc.Driver"
          } else if (jdbc.contains("oracle")) {
            "oracle.jdbc.driver.OracleDriver"
          } else {
            ""
          }
        }
      }
    
      case class Student_Info(name:String,age:Int,sno:Int,classRoom:String)
      def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("ReadDataToDf").setMaster("local[1]")
        val sc = new SparkContext(conf)
        sc.setLogLevel("WARN")
        val spark = SparkSession.builder().getOrCreate()
        import spark.implicits._
        //写入oracle数据库
        val oracle_jdbc = jdbcConnectInfo("stuInfo", "jdbc:oracle:thin:@127.0.0.1:1521/orcl", "system", "123456")
        val df = readDataBaseToDf(oracle_jdbc)
          //数据库时varchar,为了和实体类Int对应
         .withColumn("sno",expr("cast(sno as int)"))
          .as[Student_Info]
        val ageFlag:Int=23
        df.filter(s"age>=$ageFlag").sort($"age".asc).dropDuplicates("age").show()
    
        df.filter(s"age>=$ageFlag").sort($"sno".desc).dropDuplicates("age").show()
    
        val info: Student_Info = df.filter(s"age>=$ageFlag").sort($"sno".desc).dropDuplicates("age").collectAsList().get(0)
        println(info.name)
      }
    
      def readDataBaseToDf(jdbc: jdbcConnectInfo): DataFrame ={
        val spark=SparkSession.builder().getOrCreate()
        spark.read.format("jdbc")
          .option("url",jdbc.jdbc)
          .option("dbtable",jdbc.talbeName)
          .option("driver",jdbc.get_jdbc)
          .option("user", jdbc.user)
          .option("password", jdbc.password)
          .load()
      }
    }

# 三、运行结果 #

+--------+---+---+---------+
    |    name|age|sno|classRoom|
    +--------+---+---+---------+
    |xiaoMing| 23|112|      A01|
    | xiaoJun| 24|113|      A02|
    | xiaoHua| 25|114|      A03|
    +--------+---+---+---------+
    
    +--------+---+---+---------+
    |    name|age|sno|classRoom|
    +--------+---+---+---------+
    | xiaoHua| 25|114|      A03|
    |xiaoMing| 23|112|      A01|
    | xiaoJun| 24|113|      A02|
    +--------+---+---+---------+
    
    xiaoHua

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2h5al9raW5n_size_16_color_FFFFFF_t_70]: /images/20221124/97598087050b494f85704e5dab97b80f.png