sparkSQL之将数据写入不同的数据源

不念不忘少年蓝@ 2024-04-18 10:11 29阅读 0赞

object DateSourceTest {
      def main(args: Array[String]): Unit = {
        //1、创建spqrkSession
        val spark: SparkSession = SparkSession.builder().appName("w").master("local[*]").getOrCreate()
        val sc: SparkContext = spark.sparkContext
        sc.setLogLevel("WARN")
    
        //2、读取文本文件
        //val df: DataFrame = spark.read.text("") //注意：直接读取的文件返回的df没有完整的schema
        val fileRDD: RDD[String] = sc.textFile("E:\\ideal_workspace\\spark\\day01\\person.txt")
    
        //3、处理数据
        val lineRDD: RDD[Array[String]] = fileRDD.map(_.split(" "))
        //4、将RDD和样例类进行关联
        val personRDD: RDD[Person] = lineRDD.map(arr=>Person(arr(0).toInt,arr(1),arr(2).toInt))
        //将RDD转化成DF
        //注意：RDD中没有toDF方法，需要导入隐式转化
        import spark.implicits._
        val df: DataFrame = personRDD.toDF
        //注意：personRDD中有泛型信息，转化为DF的时候，DataFrame没有了泛型信息，但是通过RDD的Person泛型信息，可以反射出需要添加的Schmea信息
    
        //将DF写入不同的数据源
        //注意：不能写入普通文本，因为它有约束，是3列，而数据源只支持一列
        //json,parquet等底层都是调用了format方法
        df.write.json("E:\\ideal_workspace\\spark\\day01\\output\\json")
        df.write.parquet("E:\\ideal_workspace\\spark\\day01\\output\\parquet")
        df.write.csv("E:\\ideal_workspace\\spark\\day01\\output\\csv")
    
        var prop = new Properties()
        prop.getProperty("user","root")
        prop.setProperty("passqord","root")
        //df.write.mode(SaveMode.Overwrite).jdbc(" ","person",prop)
        println("写入成功")
      }
    
      case class Person(id:Int,name:String,age:Int)
    }