SparkSQL读写数据

2023-12-21 10:14:15

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博客

文章来源:https://blog.csdn.net/HYSliuliuliu/article/details/134983578
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。