Spark(十二)【SparkSql中数据读取和保存】

2022-03-23 22:59:59

一. 读取和保存说明

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: \t,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()

	 //通用的读

通用的读

读取mysql的数据

/**

 * @description: 测试读取mysql数据

 * @author: HaoWu

 * @create: 2020年09月11日

 */

object ReadMysqlTest {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[*]").setAppName("readMysql")

    val spark = SparkSession

      .builder()

      .config(conf)

      .getOrCreate()

    import spark.implicits._

    val ids = List(1,2,3,4).mkString("'", "','", "'")

    val resutl = spark

      .read

      .format("jdbc")

      .option("url", "jdbc:mysql://hadoop102:3306/gmall0421?useSSL=false")

      .option("user", "root")

      .option("password", "root")

      .option("query", s"select * from user_info where id in (${ids})")

      .load()

      .as[UserInfo] // df -> ds

      .rdd

      .map(userInfo => (userInfo.id, userInfo))

    resutl.collect().foreach(print)

  }

}

保存

    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()

码农公寓