Spark系列--SparkSQL(六)数据源_spark支持的数据源

前言

SparkSQL的数据源：结构化的文件（json，parquet），或者是Hive的表，或者是外部的数据库（mysql），也或者是已经存在的RDD。

Spark SQL的DataFrame接口支持多种数据源的操作。一个DataFrame可以进行RDDs方式的操作，也可以被注册为临时表。把DataFrame注册为临时表之后，就可以对该DataFrame执行SQL查询。

Spark SQL的默认数据源为Parquet格式。数据源为Parquet文件时，Spark SQL可以方便的执行所有的操作。修改配置项spark.sql.sources.default，可修改默认数据源格式。

一、通用加载/保存方法

1.1 手动指定选项

• Load是用来读取文件的时候加载文件中的数据
• Save是用来往外写文件的时候存储写出的数据

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

df.select("name", "favorite_color").write.save("namesAndFavColors.parquet")
1
2
3

当数据源格式不是parquet格式文件时，需要手动指定数据源的格式。数据源格式需要指定全名（例如：org.apache.spark.sql.parquet），如果数据源格式为内置格式，则只需要指定简称定json, parquet, jdbc, orc, libsvm, csv, text来指定数据的格式。
可以通过SparkSession提供的read.load方法用于通用加载数据，使用write和save保存数据。

val peopleDF = spark.read.format("json").load("examples/src/main/resources/people.json")


peopleDF.write.format("parquet").save("hdfs://master01:9000/namesAndAges.parquet")
1
2
3

除此之外，可以直接运行SQL在文件上：

val sqlDF = spark.sql("SELECT * FROM parquet.`hdfs://hadoop0:9000/namesAndAges.parquet`")

sqlDF.show()
1
2
3

1.2 文件保存选项

可以采用SaveMode执行存储操作，SaveMode定义了对数据的处理模式。需要注意的是，这些保存模式不使用任何锁定，不是原子操作。此外，当使用Overwrite方式执行时，在输出新数据之前原数据就已经被删除。SaveMode详细介绍如下表：

---

二、Parquet文件

Parquet是一种流行的列式存储格式，可以高效地存储具有嵌套字段的记录。

2.1 Parquet读写

Parquet格式经常在Hadoop生态圈中被使用，它也支持Spark SQL的全部数据类型。Spark SQL 提供了直接读取和存储 Parquet 格式文件的方法。

import spark.implicits._

val peopleDF = spark.read.json("examples/src/main/resources/people.json")



peopleDF.write.parquet("hdfs://hadoop0:9000/people.parquet")


val parquetFileDF = spark.read.parquet("hdfs://hadoop0:9000/people.parquet")


parquetFileDF.createOrReplaceTempView("parquetFile")


val namesDF = spark.sql("SELECT name FROM parquetFile WHERE age BETWEEN 13 AND 19")


namesDF.map(attributes => "Name: " + attributes(0)).show()

1
2
3
4
5
6
7
8
9
10
11
12
13
14

2.2 解析分区信息

对表进行分区是对数据进行优化的方式之一。在分区的表内，数据通过分区列将数据存储在不同的目录下。Parquet数据源现在能够自动发现并解析分区信息。例如，对人口数据进行分区存储，分区