SparkSQL读写数据

1.3 SparkSQL读写数据

1.3.1 数据的加载

Sparksql中加载外部的数据，使用统一的API入口，

spark.read.format(数据文件格式).load(path)

这个方式有更加清晰的简写方式，比如要加载json格式的文件

spark.read.json(path)

默认加载的文件格式为parquet

def main(args: Array[String]): Unit = {
 ? ?val spark = SparkSession.builder()
 ?  .master("local[*]")
 ?  .appName("SparkSQLLoadAndSave")
 ?  .getOrCreate()
?
 ? ?//加载数据 默认的加载文件的格式为parquet
 ? ?var pdf = spark.read.format("json").load("file:///E:/data/spark/sql/people.json")
 ? ?//简写方式
 ? ?pdf = spark.read.json("file:///E:/data/spark/sql/people.json")
 ? ?//parquet
 ? ?pdf = spark.read.parquet("file:///E:/data/spark/sql/users.parquet")
 ? ?//text 加载普通的文本文件，只能解析成一列
 ? ?pdf = spark.read.text("file:///E:/data/spark/sql/dailykey.txt")
 ? ?//csv 普通的文本文件，但是每一列之间是,作为分隔符
 ? ?pdf = spark.read.csv("file:///E:/data/spark/sql/province.csv")
 ?  .toDF("pid", "province", "code", "cid")//重新根据需要命名列名 数据类型全部是字符串
 ? ?//orc 是rc文件格式的升级版本
 ? ?pdf = spark.read.orc("file:///E:/data/spark/sql/student.orc")
 ? ?//jdbc
 ? ?val url = "jdbc:mysql://localhost:3306/test"
 ? ?val table = "wordcount"
 ? ?val properties = new Properties()
 ? ?properties.put("user", "bigdata")
 ? ?properties.put("password", "sorry")
 ? ?pdf = spark.read.jdbc(url, table, properties)
?
 ? ?pdf.printSchema()
 ? ?pdf.show()
 ? ?// 关闭
 ? ?spark.stop()
}

1.3.2 ?数据的落地

    SparkSQL对数据的落地保存使用api为：spark.write.save()，需要指定数据的落地格式，因为和read的默认格式一样，save的默认格式也是parquet，需要在write和save之间指定具体的格式format(format)
?
    同样也有简写方式：spark.write.json/parquet等等

 ? ?def main(args: Array[String]): Unit = {
 ? ? ? ?val spark = SparkSession.builder()
 ? ? ? ? ? ? ? ? ?  .master("local[*]")
 ? ? ? ? ? ? ? ? ?  .appName("SparkSQLLoadAndSave")
 ? ? ? ? ? ? ? ? ?  .getOrCreate()
?
 ? ? ? ?val df = spark.read.orc("file:///E:/data/spark/sql/student.orc")
 ? ? ? ?/*
 ? ? ? ? ? ?数据的落地
 ? ? ? ? ? ? ? 默认的存储格式为parquet，同时基于snappy压缩方式存储
 ? ? ? ? ? ?落地的保存方式SaveMode
 ? ? ? ? ? ? ? ?ErrorIfExists:目录存在保存，默认的格式
 ? ? ? ? ? ? ? ?Append：在原有的基础之上追加
 ? ? ? ? ? ? ? ?Ignore：忽略，如果目录存在则忽略，不存在则创建
 ? ? ? ? ? ? ? ?Overwrite：覆盖(删除并重建)
 ? ? ? ? */
// ? ? ?  df.write.format("json").mode(SaveMode.Overwrite).save("file:///E:/data/spark/sql/stu")
 ? ? ? ?val url = "jdbc:mysql://localhost:3306/test"
 ? ? ? ?val table = "student"
 ? ? ? ?val properties = new Properties()
 ? ? ? ?properties.put("user", "bigdata")
 ? ? ? ?properties.put("password", "sorry")
 ? ? ? ?df.write.mode(SaveMode.Append).jdbc(url, table, properties)
 ? ? ? ?spark.stop()
 ?  }

Guff_hys_python数据结构,大数据开发学习,python实训项目-CSDN博客