一. 读取和保存说明
SparkSQL提供了通用的保存数据和数据加载的方式,还提供了专用的方式
读取:通用和专用
保存
保存有四种模式:
默认: error : 输出目录存在就报错
append: 向输出目录追加
overwrite : 覆盖写
ignore: 忽略,不写
二. 数据格式
1. Parquet
Spark SQL的默认数据源为Parquet格式。Parquet是一种能够有效存储嵌套数据的列式存储格式。
数据源为Parquet文件时,Spark SQL可以方便的执行所有的操作,不需要使用format。修改配置项spark.sql.sources.default,可修改默认数据源格式。
读取
val df = spark.read.load("examples/src/main/resources/users.parquet")
保存
//读取json文件格式
var df = spark.read.json("/opt/module/data/input/people.json")
//保存为parquet格式
df.write.mode("append").save("/opt/module/data/output")
2. Json
Spark SQL 能够自动推测JSON数据集的结构,并将它加载为一个Dataset[Row]. 可以通过SparkSession.read.json()去加载JSON 文件。
注意:Spark读取的JSON文件不是传统的JSON文件,每一行都应该是一个JSON串。
数据格式:employees.json
{"name":"Michael"}
{"name":"Andy", "age":30}
1)导入隐式转换
import spark.implicits._
2)读取Json文件
//专用的读取
val df1: DataFrame = sparkSession.read.json("input/employees.json")
//通用读取
val df: DataFrame = sparkSession.read.format("json").load("input/employees.json")
3)保存为Json文件
//导隐式包,转为DataSet
import sparkSession.implicits.
val ds: Dataset[Emp] = rdd.toDS()
ds.write.mode("overwrite")json("output/emp.json")
3. CSV
CSV: 逗号作为字段分割符的文件
tsv: ,tab作为字段分割符的文件
读取
// 通用的读取
val df: DataFrame = sparkSession.read.format("csv").load("input/person.csv")
// 专用的读
val df1: DataFrame = sparkSession.read.csv("input/person.csv")
保存
CSV的参数可以到DataFrameReader 609行查看
//DataFrame
df1.write.option("sep",",").mode("overwrite").csv("output/csv")
4. Mysql
读取
val props = new Properties()
/*
JDBC中能写什么参数,参考 JDBCOptions 223行
*/
props.put("user","root")
props.put("password","root")
//库名
val df: DataFrame = sparkSession.read.jdbc("jdbc:mysql://localhost:3306/spark_test", "tbl_user", props)
// 全表查询 只显示前N条
df.show()
//指定查询
df.createTempView("user")
sparkSession.sql("select * from user where id > 5").show()
保存
val list = List(Emp("jack", 2222.22), Emp("jack1", 3222.22), Emp("jack2", 4222.22))
val rdd: RDD[Emp] = sparkSession.sparkContext.makeRDD(list, 1)
//导入隐式包
import sparkSession.implicits._
val ds: Dataset[Emp] = rdd.toDS()
val props = new Properties()
props.put("user","root")
props.put("password","root")
// 表名可以是已经存在的表t1,也可以是一张新表t1(用的多)
//专用的写
ds.write.jdbc("jdbc:mysql://localhost:3306/0508","t1",props)
// 通用的写
ds.write.
option("url","jdbc:mysql://localhost:3306/库名")
//表名
.option("dbtable","t2")
.option("user","root")
.option("password","root")
.mode("append")
.format("jdbc").save()