09.SparkSQL数据的加载和保存

Leefs 2021-07-16 PM 2005℃ 0条

# 09.SparkSQL数据的加载和保存

### 一、通用的加载和保存方式

SparkSQL 提供了通用的保存数据和数据加载的方式。这里的通用指的是使用相同的 API，根据不同的参数读取和保存不同格式的数据，SparkSQL 默认读取和保存的文件格式 为 parquet

#### 1.1 加载数据

`spark.read.load`是加载数据的通用方法

![09.SparkSQL数据的加载和保存01.jpg](https://lilinchao.com/usr/uploads/2021/07/3967548489.jpg)

下方都是spark.read支持的加载数据方式

如果读取不同格式的数据，可以对不同的数据格式进行设定

**加载数据常用命令**

```scala
scala> spark.read.format("…")[.option("…")].load("…")
```

**说明：**

+ format("…")：指定加载的数据类型，包括"csv"、"jdbc"、"json"、"orc"、"parquet"和 "textFile"。
+ load("…")：在"csv"、"jdbc"、"json"、"orc"、"parquet"和"textFile"格式下需要传入加载数据的路径。

+ option("…")：在"jdbc"格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable

**通过在文档中直接查询数据**

我们前面都是使用 read API 先把文件加载到 DataFrame 然后再查询，其实，我们也可以直接在文件上进行查询:

**文件格式.`文件路径`**

**命令**

```scala
scala>spark.sql("select * from json.`/opt/module/data/user.json`").show
```

#### 1.2 保存数据

`df.write.save` 是保存数据的通用方法

```scala
scala>df.write.
csv jdbc json orc parquet textFile… …
```

**保存数据命令**

```scala
scala>df.write.format("…")[.option("…")].save("…")
```

**说明：**

+ format("…")：指定保存的数据类型，包括"csv"、"jdbc"、"json"、"orc"、"parquet"和 "textFile"。
+ save ("…")：在"csv"、"orc"、"parquet"和"textFile"格式下需要传入保存数据的路径。
+ option("…")：在"jdbc"格式下需要传入 JDBC 相应参数，url、user、password 和 dbtable

保存操作可以使用 SaveMode, 用来指明如何处理数据，使用 mode()方法来设置。

有一点很重要: 这些 SaveMode 都是没有加锁的, 也不是原子操作。

**SaveMode 是一个枚举类，其中的常量包括：**

| Scala/Java                      | Any Language     | Meaning                    |
| ------------------------------- | ---------------- | -------------------------- |
| SaveMode.ErrorIfExists(default) | "error"(default) | 如果文件已经存在则抛出异常 |
| SaveMode.Append                 | "append"         | 如果文件已经存在则追加     |
| SaveMode.Overwrite              | "overwrite"      | 如果文件已经存在则覆盖     |
| SaveMode.Ignore                 | "ignore"         | 如果文件已经存在则忽略     |

**命令**

```scala
df.write.mode("append").json("/opt/module/data/output")
```

### 二、操作文件的三种方式介绍

#### 2.1 Parquet

Spark SQL 的默认数据源为 Parquet 格式。Parquet 是一种能够有效存储嵌套数据的列式 存储格式。

数据源为 Parquet 文件时，Spark SQL 可以方便的执行所有的操作，不需要使用 format。

修改配置项 `spark.sql.sources.default`，可修改默认数据源格式。

+ 加载数据

```scala
scala> val df = spark.read.load("examples/src/main/resources/users.parquet")
scala> df.show
```

+ 保存数据

```scala
scala> var df = spark.read.json("/opt/module/data/input/people.json")
//保存为 parquet 格式
scala> df.write.mode("append").save("/opt/module/data/output")
```

#### 2.2 JSON

Spark SQL 能够自动推测 JSON 数据集的结构，并将它加载为一个 Dataset[Row]. 可以通过 `SparkSession.read.json()`去加载 JSON 文件。

**注意：Spark 读取的 JSON 文件不是传统的 JSON 文件，每一行都应该是一个 JSON 串。**

因为Spark读取文件是按行读取的。

格式如下：

```json
{"name":"Michael"}
{"name":"Andy"， "age":30}
[{"name":"Justin"， "age":19},{"name":"Justin"， "age":19}]
```

**（1）导入隐式转换**

```scala
import spark.implicits._
```

**（2）加载 JSON 文件**

```scala
val path = "/opt/module/spark-local/people.json"
val peopleDF = spark.read.json(path)
```

**（3）创建临时表**

```scala
peopleDF.createOrReplaceTempView("people")
```

**（4）数据查询**

```scala
val teenagerNamesDF = spark.sql("SELECT name FROM people WHERE age BETWEEN 13 
AND 19")
teenagerNamesDF.show()
```

#### 2.3 CSV

SparkSQL读取csv可以根据csv文件的第一行作为header自动推导出列名或schema,也可以通过手动的方式指定schema,自动推导读取csv时需要指定option参数

**命令**

```scala
spark.read()
	  //自动推断列类型
	   .option("inferSchema", "true")
	   //指定一个指示空值的字符串
	   .option("nullvalue", "?")
	   //当设置为 true 时，第一行文件将被用来命名列，而不包含在数据中
	   .option("header", "true")
	   .csv("/home/cry/myStudyData/userList.csv");
```

标签: Spark, Spark SQL

非特殊说明，本博所有文章均为博主原创。

如若转载，请注明出处：https://lilinchao.com/archives/1356.html

上一篇 08.UDF和UDAF函数介绍

下一篇 10.【转载】SparkSQL之Join实现介绍

09.SparkSQL数据的加载和保存

评论已关闭

栏目分类

标签云

友情链接申请

09.SparkSQL数据的加载和保存

评论已关闭

 栏目分类

标签云

友情链接申请

栏目分类

标签云

友情链接申请