来源 |  Learning Spark Lightning-Fast Data Analytics，Second Edition

作者 | Damji，et al.

翻译 | 吴邪 大数据4年从业经验，目前就职于广州一家互联网公司，负责大数据基础平台自研、离线计算&实时计算研究

校对 | gongyouliu

编辑 | auroral-L

全文共14503字，预计阅读时间80分钟。

第四章  Spark SQL 和 DataFrames：

内置数据源简介

1.  在 Spark 应用程序中使用 Spark SQL

1.1  基本查询示例

2.  SQL 表和视图

2.1  托管与非托管表（Managed Versus UnmanagedTables）

2.2  创建 SQL 数据库和表

2.3  创建视图

2.4  查看元数据

2.5  缓存 SQL 表

2.6  读取表写入DataFrame

3.  DataFrames 和 SQL 表的数据源

3.1  DataFrameReader

3.2  DataFrameWriter

3.3  Parquet

3.4  JSON格式

3.5  CSV

3.6  Avro

3.7  ORC

3.8  Images

3.9  二进制文件

4.  总结

在上一章中，我们解释了Spark结构化的演变及其合理性。特别是，我们讨论了Spark SQL引擎如何为高级DataFrame和Dataset API提供统一的接口。现在，我们将继续讨论DataFrame，并探讨其与Spark SQL的交互。

本章和下一章还将探讨Spark SQL如何与图4-1中所示的一些外部组件交互。

特别是Spark SQL：

• 提供了用于构建我们在第3章中探讨的高级结构化API的引擎。

• 可以读写各种格式的结构化数据（例如，JSON，Hive表，Parquet，Avro，ORC，CSV）。

• 使你可以使用JDBC / ODBC连接器从Tableau，Power BI，Talend等外部商业智能（BI）数据源或从Postgres和MySQL等RDBMS来查询数据。

• 提供与Spark应用程序中存储为数据库中表或视图的结构化数据进行交互的编程接口。

• 提供一个交互式shell程序，用于对结构化数据执行SQL查询。

• 支持ANSI SQL：2003标准的命令和HiveQL。

让我们从如何在Spark应用程序中使用Spark SQL开始入手。

1.  在Spark应用程序中使用Spark SQL

在Spark2.0中引入的SparkSession为使用结构化API编写Spark提供了一个统一的切入点。你可以使用SparkSession来调用Spark的功能：只需导入类并在代码中创建一个实例。

在SparkSession上使用sql()方法实例化spark执行SQL查询，例如spark.sql("SELECT * FROM myTableName")。以spark.sql这种方式执行的所有查询结果都会返回一个DataFrame，如果你需要，可以在该DataFrame上执行进一步的Spark操作----我们在第3章中探讨的那些操作以及在本章和下一章中将学到的方法。

1.1  基本查询示例

在本节中，我们将通过几个示例查询有关航空公司的航班准点性和航班延误原因的数据集，该数据集包含有关美国航班的数据，包括日期，延误，距离，始发地和目的地。它以CSV文件的形式提供，超过一百万条记录。通过定义schema，我们将数据读取到DataFrame并将该DataFrame注册为一个临时视图（稍后将在临时视图中进行更多介绍），以便我们可以使用SQL查询它。

代码段中提供了查询示例，而本书的GitHub repo中提供了包含此处介绍的所有代码的Python和Scala笔记（notebook）。这些示例将使你了解如何通过spark.sql编程接口在Spark应用程序中使用SQL。与声明性风格的DataFrame API相似，此接口允许你在Spark应用程序中查询结构化数据。

通常，在Standalone模式下的Spark应用程序中，你可以手动创建一个SparkSession实例，如以下示例所示。但是，在Spark Shell（或Databricks 笔记）中，默认为你创建了SparkSession，并赋值给变量spark，你可以通过spark变量进行访问。

接下来让我们开始将数据集读取到一个临时视图中：

// In Scala
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder.appName("SparkSQLExampleApp").getOrCreate()
// Path to data set
val csvFile="/databricks-datasets/learning-spark-v2/flights/departuredelays.csv"
// Read and create a temporary view
// Infer schema (note that for larger files you may want to specify the schema)
val df = spark.read.format("csv").option("inferSchema", "true").option("header", "true").load(csvFile)
// Create a temporary view
df.createOrReplaceTempView("us_delay_flights_tbl")

# In Python
from pyspark.sql import SparkSession
# Create a SparkSession
spark = (SparkSession.builder.appName("SparkSQLExampleApp").getOrCreate())
# Path to data set
csv_file = "/databricks-datasets/learning-spark-v2/flights/departuredelays.csv"
# Read and create a temporary view
# Infer schema (note that for larger files you
# may want to specify the schema)
df = (spark.read.format("csv").option("inferSchema", "true").option("header", "true").load(csv_file))
df.createOrReplaceTempView("us_delay_flights_tbl")

如果要指定schema，则可以使用DDL格式的字符串。例如：
// In Scala
val schema = "date STRING, delay INT, distance INT, origin STRING, destination STRING"# In Python
schema = "`date` STRING, `delay` INT, `distance` INT, `origin` STRING, `destination` STRING"

现在我们已经有了一个临时视图，我们可以使用Spark SQL执行SQL查询。这些查询与你可能针对MySQL或PostgreSQL数据库中的SQL表执行的查询没有什么不同。这里的重点是表明Spark SQL提供了一个符合ANSI：2003的SQL接口，并演示了SQL与DataFrames之间的相互可操作性。

美国航班延误数据集有五列：

• date列包含类似的字符串02190925。转换后，它映射到02-19 09:25 am。

• delay列以分钟为单位给出了计划的起飞时间与实际起飞时间之间的延迟。提早出发显示负数。

• distance列给出了从始发机场到目的地机场的距离（以英里为单位）。

• origin列包含始发国际航空运输协会机场代码。

• destination列包含目的地国际航空运输协会机场代码。

考虑到这一点，让我们尝试针对此数据集进行一些示例查询。

首先，我们将查找距离大于1000英里的所有航班：

spark.sql("""SELECT distance, origin, destination
FROM us_delay_flights_tbl WHERE distance > 1000
ORDER BY distance DESC""").show(10)
+--------+------+-----------+
|distance|origin|destination|
+--------+------+-----------+
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
+--------+------+-----------+
only showing top 10 rows

结果显示，所有最长的航班都在檀香山（HNL）和纽约（JFK）之间。接下来，我们将查找出旧金山（SFO）和芝加哥（ORD）之间延迟超过两个小时的所有航班：

spark.sql("""SELECT date, delay, origin, destination
FROM us_delay_flights_tbl
WHERE delay > 120 AND ORIGIN = 'SFO' AND DESTINATION = 'ORD'
ORDER by delay DESC""").show(10)
+--------+-----+------+-----------+
|date |delay|origin|destination|
+--------+-----+------+-----------+
|02190925|1638 |SFO |ORD |
|01031755|396 |SFO |ORD |
|01022330|326 |SFO |ORD |
|01051205|320 |SFO |ORD |
|01190925|297 |SFO |ORD |
|02171115|296 |SFO |ORD |
|01071040|279 |SFO |ORD |
|01051550|274 |SFO |ORD |
|03120730|266 |SFO |ORD |
|01261104|258 |SFO |ORD |
+--------+-----+------+-----------+
only showing top 10 rows

看来这两个城市之间在不同的日期有很多明显的航班延误。（作为练习，将date列转换为可读的格式，并找出这些延迟最常见的日期或月份。思考这些延迟与冬季或假日有关吗？）

让我们尝试一个更复杂的查询，其中在SQL语句中使用CASE子句。在这个示例中，我们要标记所有美国航班，无论其始发地和目的地如何，以表明其经历的延误：超长延误（> 6小时），长延误（2–6小时）等。将这些人类可读的标签添加到名为的新列中Flight_Delays：

spark.sql("""SELECT delay, origin, destination,
CASE
WHEN delay > 360 THEN 'Very Long Delays'
WHEN delay > 120 AND delay < 360 THEN 'Long Delays'
WHEN delay > 60 AND delay < 120 THEN 'Short Delays'
WHEN delay > 0 and delay < 60 THEN 'Tolerable Delays'
WHEN delay = 0 THEN 'No Delays'
ELSE 'Early'
END AS Flight_Delays
FROM us_delay_flights_tbl
ORDER BY origin, delay DESC""").show(10)
+-----+------+-----------+-------------+
|delay|origin|destination|Flight_Delays|
+-----+------+-----------+-------------+
|333 |ABE |ATL |Long Delays |
|305 |ABE |ATL |Long Delays |
|275 |ABE |ATL |Long Delays |
|257 |ABE |ATL |Long Delays |
|247 |ABE |DTW |Long Delays |
|247 |ABE |ATL |Long Delays |
|219 |ABE |ORD |Long Delays |
|211 |ABE |ATL |Long Delays |
|197 |ABE |DTW |Long Delays |
|192 |ABE |ORD |Long Delays |
+-----+------+-----------+-------------+
only showing top 10 rows

与DataFrame和Dataset API一样，通过spark.sql接口，你可以执行常见的数据分析操作，如我们在上一章中探讨的那样。该计算经过Spark SQL引擎相同的流程（见第三章中“Catalyst优化器”了解详细信息），最终得到相同的结果。

前面的所有三个SQL查询都可以用等效的DataFrame API查询表示。例如，第一个查询可以在Python DataFrame API中表示为：

# In Python
from pyspark.sql.functions import col, desc
(df.select("distance", "origin", "destination").where(col("distance") > 1000).orderBy(desc("distance"))).show(10)
# Or
(df.select("distance", "origin", "destination").where("distance > 1000").orderBy("distance", ascending=False).show(10))

这将产生与SQL查询相同的结果：

+--------+------+-----------+
|distance|origin|destination|
+--------+------+-----------+
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
|4330 |HNL |JFK |
+--------+------+-----------+
only showing top 10 rows

作为练习，请尝试将其他两个SQL查询转换为DataFrame API的形式。

如这些示例所示，使用Spark SQL接口查询数据类似于用常规SQL查询关系数据库表。尽管查询是在SQL中进行的，但你会感觉到与第3章中遇到的DataFrame API操作在可读性和语义上的相似之处，我们将在下一章中进一步探讨。

为了使你能够如前面的示例中所示查询结构化数据，Spark在内存和磁盘中创建和管理视图和表的所有复杂性。这就引出了我们的下一个主题：如何创建和管理表和视图。

2.  SQL表和视图

表中包含数据。与Spark中的每个表相关联的是其相关的元数据，它是有关表及其数据的信息：数据结构，描述，表名，数据库名，列名，分区，实际数据所在的物理位置等。所有这些存储在metastore中。

默认情况下，Spark使用位于/user/hive/warehouse的Apache Hive Metastore来保留关于表的所有元数据，而不是为Spark表提供单独的元存储。但是，你可以通过将Spark config的配置spark.sql.warehouse.dir设置为另一个目录来更改默认位置，该位置可以设置为本地目录或外部分布式存储。

2.1  托管与非托管表（Managed Versus UnmanagedTables）

Spark允许你创建两种类型的表：托管表和非托管表。对于托管表，Spark同时管理文件存储中的元数据和数据。这可以是本地文件系统，HDFS或对象存储，例如Amazon S3或Azure Blob。对于非托管表，Spark仅管理元数据，而你自己在外部数据源（例如Cassandra）中管理数据。

对于托管表，由于Spark可以管理所有内容，因此SQL命令如DROP TABLE table_name会将元数据和数据一起删除。对于非托管表，同一命令将仅删除元数据，而不删除实际数据。在下一节中，我们将介绍一些有关如何创建托管表和非托管表的示例。

2.2  创建SQL数据库和表

表驻留在数据库中。默认情况下，Spark在default数据库下创建表。要创建自己的数据库名称，可以从Spark应用程序或笔记执行SQL命令。使用美国航班延误数据集，让我们创建一个托管表和一个非托管表。首先，我们将创建一个名为learn_spark_db的数据库，并告诉Spark我们要使用该数据库：

// In Scala/Python
spark.sql("CREATE DATABASE learn_spark_db")
spark.sql("USE learn_spark_db")

从这一点来看，我们在应用程序中执行的用于创建表的任何命令都会作用于learn_spark_db数据库下所有的表。

创建托管表

要在learn_spark_db数据库中创建托管表，可以执行如下SQL查询：

// In Scala/Python
spark.sql("CREATE TABLE managed_us_delay_flights_tbl (date STRING, delay INT, distance INT, origin STRING, destination STRING)")

你也可以使用DataFrame API进行相同的操作，如下所示：

# In Python
# Path to our US flight delays CSV file
csv_file = "/databricks-datasets/learning-spark-v2/flights/departuredelays.csv"
# Schema as defined in the preceding example
schema="date STRING, delay INT, distance INT, origin STRING, destination STRING"
flights_df = spark.read.csv(csv_file, schema=schema)
flights_df.write.saveAsTable("managed_us_delay_flights_tbl")

这两个语句最后都能在learn_spark_db数据库中创建us_delay_flights_tbl托管表。

创建非托管表

相比之下，你可以从自己支持Spark应用读取的数据源（例如Parquet，CSV或JSON文件）创建，要从数据源（例如CSV文件）创建非托管表，请使用如下SQL：

spark.sql("""CREATE TABLE us_delay_flights_tbl(date STRING, delay INT,
distance INT, origin STRING, destination STRING)
USING csv OPTIONS (PATH
'/databricks-datasets/learning-spark-v2/flights/departuredelays.csv')""")

在DataFrame API中使用：

(flights_df
.write
.option("path", "/tmp/data/us_flights_delay")
.saveAsTable("us_delay_flights_tbl"))

为了使你能够浏览这些示例，我们创建了Python和Scala示例笔记，你可以在本书的GitHub repo中找到这些笔记。

2.3  创建视图

除了创建表之外，Spark还可在现有表之上创建视图。视图可以是全局视图（SparkSession在给定集群的所有节点上可见）或会话范围视图（仅对单个SparkSession可见），并且它们是临时的：会随着Spark应用程序终止而被回收。创建视图的语法与在数据库中创建表的语法相似。创建视图后，就可以像查询表一样对其进行查询。视图和表之间的区别在于，视图实际上并不保存数据。Spark应用程序终止后，表仍然存在，但视图会被回收。你可以使用SQL从现有表创建视图。例如，如果你只希望使用纽约（JFK）和旧金山（SFO）的始发机场处理美国航班延误数据集的子集，则以下查询将创建仅由该切片组成的全局临时视图和临时视图表：

-- In SQL
CREATE OR REPLACE GLOBAL TEMP VIEW us_origin_airport_SFO_global_tmp_view AS
SELECT date, delay, origin, destination from us_delay_flights_tbl WHERE
origin = 'SFO';
CREATE OR REPLACE TEMP VIEW us_origin_airport_JFK_tmp_view AS
SELECT date, delay, origin, destination from us_delay_flights_tbl WHERE
origin = 'JFK'

你也可以使用DataFrame API完成相同的操作，如下所示：

# In Python
df_sfo = spark.sql("SELECT date, delay, origin, destination FROM
us_delay_flights_tbl WHERE origin = 'SFO'")
df_jfk = spark.sql("SELECT date, delay, origin, destination FROM
us_delay_flights_tbl WHERE origin = 'JFK'")
# Create a temporary and global temporary view
df_sfo.createOrReplaceGlobalTempView("us_origin_airport_SFO_global_tmp_view")
df_jfk.createOrReplaceTempView("us_origin_airport_JFK_tmp_view")

一旦创建了这些视图，就可以像对表一样对它们执行查询。请记住，访问全局临时视图时，必须使用前缀，如global_temp.<view_name>，因为Spark在名为的全局临时数据库（global_temp）中创建全局临时视图。

-- In SQL
SELECT * FROM global_temp.us_origin_airport_SFO_global_tmp_view

相比之下，你可以访问不带global_temp前缀的普通临时视图。

-- In SQL
SELECT * FROM us_origin_airport_JFK_tmp_view

// In Scala/Python
spark.read.table("us_origin_airport_JFK_tmp_view")
// Or
spark.sql("SELECT * FROM us_origin_airport_JFK_tmp_view")

你也可以像删除表一样删除视图：

-- In SQL
DROP VIEW IF EXISTS us_origin_airport_SFO_global_tmp_view;
DROP VIEW IF EXISTS us_origin_airport_JFK_tmp_view
// In Scala/Python
spark.catalog.dropGlobalTempView("us_origin_airport_SFO_global_tmp_view")
spark.catalog.dropTempView("us_origin_airport_JFK_tmp_view")

临时视图与全局临时视图

临时视图与全局临时视图之间的差异很微妙，这可能是新加入Spark的开发人员可能有困惑的地方。临时视图绑定到Spark应用程序中的单个Spark会话。相比之下，在Spark应用程序中的多个Spark会话可以看到全局临时视图。是的，你可以在单个Spark应用程序中创建多个SparkSession，这是很方便的，例如，当你要访问（并合并）来自两个不共享同一个Hive MetaStore配置的不同Spark会话的数据时可以这样做。

2.4  查看元数据

如前所述，Spark管理与每个托管或非托管表关联的元数据。这是Spark SQL中用于存储元数据的高级抽象Catalog的功能。Catalog是在Spark 2.x的扩展功能并使用了新的公共方法，使你能够检查与数据库、表和视图关联的元数据。Spark 3.0将其扩展为使用外部catalog（我们将在第12章中进行简要讨论）。例如，在Spark应用程序中，创建SparkSession之后赋值成变量spark，你可以通过以下方法访问所有存储的元数据：

// In Scala/Python
spark.catalog.listDatabases()
spark.catalog.listTables()
spark.catalog.listColumns("us_delay_flights_tbl")

从本书的GitHub仓库中导入笔记，然后尝试一下。

2.5  缓存SQL表

尽管我们将在下一章讨论表缓存策略，但是值得一提的是，像DataFrames一样，你可以缓存SQL表和视图和释放SQL表和视图缓存。在Spark 3.0中，除了其他选项之外，你还可以将表指定为LAZY，这意味着该表仅应在首次使用时进行缓存，而不是立即进行缓存：

-- In SQL
CACHE [LAZY] TABLE <table-name>
UNCACHE TABLE <table-name>

2.6  读取表写入DataFrame

通常，数据工程师会在其常规数据提取和ETL流程中建立数据管道。它们使用清理后的数据填充Spark SQL数据库和表，以供下游应用程序使用。

假设你已经可以使用现有的数据库learn_spark_db和表us_delay_flights_tbl。无需读取外部JSON文件，那么你只需使用SQL查询表并将返回的结果分配给DataFrame即可：

// In Scala
val usFlightsDF = spark.sql("SELECT * FROM us_delay_flights_tbl")
val usFlightsDF2 = spark.table("us_delay_flights_tbl")
# In Python
us_flights_df = spark.sql("SELECT * FROM us_delay_flights_tbl")
us_flights_df2 = spark.table("us_delay_flights_tbl")

现在，你已经从现有的Spark SQL表中生成了一个清洗过的DataFrame了。你还可以使用Spark的内置数据源读取其他格式的数据，从而使你可以灵活地与各种常见文件格式进行交互。

3.  DataFrames和SQL表的数据源

如图4-1所示，Spark SQL提供了一个到各种数据源的接口。它还提供了一组使用Data Sources API在这些数据源之间读写数据的常用方法。

在本节中，我们将介绍一些内置数据源，可用的文件格式以及加载和写入数据的方式，以及与这些数据源有关的特定选项。但首先，让我们仔细研究两个高级数据源API结构，它们决定了你与不同数据源进行交互的方式：DataFrameReader和DataFrameWriter。

3.1  DataFrameReader

DataFrameReader是用于将数据从数据源读取到DataFrame的核心构造。它具有定义的格式和推荐的使用模式：

DataFrameReader.format(args).option("key", "value").schema(args).load()

这种类型的将方法串联在一起的做法在Spark中很常见，并且易于阅读。在探索通用数据分析模式时，我们在第3章中已经看到了它。请注意，你只能通过SparkSession实例访问DataFrameReader。也就是说，你无法创建DataFrameReader的实例。要获取实例句柄，请使用：

SparkSession.read
// or
SparkSession.readStream

当从静态数据源读取到DataFrame的句柄的同时，返回DataFrameReader，读取流返回一个要从数据流源读取的实例。（我们将在书的后面介绍结构化流）。

指向DataFrameReader的每个公共方法的参数都采用不同的值。表4-1枚举了这些参数，以及支持的参数的一个子集。

虽然我们不会列举参数和选项的所有不同组合，Python，Scala，R和Java的文档提供了建议和指导。不过，值得举几个例子：

// In Scala
// Use Parquet
val file = """/databricks-datasets/learning-spark-v2/flights/summary-
data/parquet/2010-summary.parquet"""
val df = spark.read.format("parquet").load(file)
// Use Parquet; you can omit format("parquet") if you wish as it's the default
val df2 = spark.read.load(file)
// Use CSV
val df3 = spark.read.format("csv")
.option("inferSchema", "true")
.option("header", "true")
.option("mode", "PERMISSIVE")
.load("/databricks-datasets/learning-spark-v2/flights/summary-data/csv/*")
// Use JSON
val df4 = spark.read.format("json")
.load("/databricks-datasets/learning-spark-v2/flights/summary-data/json/*")

通常，从静态Parquet数据源读取数据时不需要任何schema——Parquet元数据通常包含该schema，因此可以对其进行推断。但是，对于流数据源，你将必须提供一个schema。（我们将在第8章中介绍从流数据源进行读取。）

Parquet是Spark的默认数据源，因为它高效，使用列存储并采用快速压缩算法。当我们更深入地介绍Catalyst优化器时，你将在以后看到其他好处（例如，列式下推）。

3.2  DataFrameWriter

DataFrameWriter进行与之相反的操作：将数据保存或写入到指定的内置数据源。不同于DataFrameReader，你不是从SparkSession进行访问而是从要保存的DataFrame访问其实例。它有一些推荐的使用模式：

DataFrameWriter.format(args)
.option(args)
.bucketBy(args)
.partitionBy(args)
.save(path)

要获取实例句柄，请使用：

DataFrame.write
// or
DataFrame.writeStream

DataFrameWriter中每个方法的参数也采用不同的值。我们在表4-2中列出了这些内容，并提供了一部分受支持的参数。

这是一个简短的示例代码段，用于说明方法和参数的使用：

// In Scala
// Use JSON
val location = ...
df.write.format("json").mode("overwrite").save(location)

3.3  Parquet

我们将从Parquet开始研究数据源，因为它是Spark中的默认数据源。Parquet是许多大数据处理框架和平台所支持和广泛使用的一种开源列式文件格式，可提供许多I/O优化（例如压缩，可节省存储空间并允许快速访问数据列）。

由于其效率和这些优化，我们建议在转换和清洗数据后，将DataFrame以Parquet格式保存以供下游使用。（Parquet也是Delta Lake的默认表打开格式，我们将在第9章中介绍。）

将PARQUET文件读入DataFrame

Parquet文件存储在目录结构中，该目录结构包含数据文件，元数据，许多压缩文件和某些状态文件。页脚中的元数据包含文件格式，数据结构和列数据（例如路径等）。

例如，Parquet文件中的目录可能包含以下文件集：

_SUCCESS
_committed_1799640464332036264
_started_1799640464332036264
part-00000-tid-1799640464332036264-91273258-d7ef-4dc7-<...>-c000.snappy.parquet

目录中可能存在大量的part-XXXX压缩文件（此处显示的名称已缩短，以适合页面显示）。

要将Parquet文件读入DataFrame，只需指定格式和路径：

// In Scala
val file = """/databricks-datasets/learning-spark-v2/flights/summary-data/
parquet/2010-summary.parquet/"""
val df = spark.read.format("parquet").load(file)
# In Python
file = """/databricks-datasets/learning-spark-v2/flights/summary-data/parquet/
2010-summary.parquet/"""
df = spark.read.format("parquet").load(file)

除非你从流数据源读取数据，否则无需提供schema，因为Parquet会将其保存为元数据的一部分。

将PARQUET文件读入SPARK SQL表

除了将Parquet文件读入Spark DataFrame之外，你还可以创建Spark SQL非托管表或直接使用SQL查看。

-- In SQL
CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tbl
USING parquet
OPTIONS (
path "/databricks-datasets/learning-spark-v2/flights/summary-data/parquet/
2010-summary.parquet/" )

创建表或视图后，你可以使用SQL将数据读入DataFrame中，如我们在前面的示例中所看到的：

// In Scala
spark.sql("SELECT * FROM us_delay_flights_tbl").show()
# In Python
spark.sql("SELECT * FROM us_delay_flights_tbl").show()

这两个操作都返回相同的结果：

+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States |Romania |1 |
|United States |Ireland |264 |
|United States |India |69 |
|Egypt |United States |24 |
|Equatorial Guinea|United States |1 |
|United States |Singapore |25 |
|United States |Grenada |54 |
|Costa Rica |United States |477 |
|Senegal |United States |29 |
|United States |Marshall Islands |44 |
+-----------------+-------------------+-----+
only showing top 10 rows

将DATAFRAMES写入PARQUET文件

将DataFrame写入或保存为表或文件是Spark中的常见操作。要编写DataFrame，你只需使用DataFrameWriter本章前面概述的方法和参数，并提供将Parquet文件要保存的位置。例如：

// In Scala
df.write.format("parquet")
.mode("overwrite")
.option("compression", "snappy")
.save("/tmp/data/parquet/df_parquet")
# In Python
(df.write.format("parquet")
.mode("overwrite")
.option("compression", "snappy")
.save("/tmp/data/parquet/df_parquet"))

回想一下，Parquet是默认文件格式。如果不包括该format()方法，则DataFrame仍将另存为Parquet文件。

这将在指定的路径上创建一组紧凑和压缩的Parquet文件。由于我们在这里使用snappy作为压缩选项，因此我们将拥有snappy压缩文件。为简便起见，本示例仅生成一个文件；仅此而已。通常，可能会创建大约十二个文件：

将DATAFRAMES写入SPARK SQL表

将DataFrame写入SQL表就像写入文件一样容易，只需使用saveAsTable()即可，而不是save()。这将创建一个称为us_delay_flights_tbl：的托管表：

// In Scala
df.write
.mode("overwrite")
.saveAsTable("us_delay_flights_tbl")
# In Python
(df.write
.mode("overwrite")
.saveAsTable("us_delay_flights_tbl"))

综上所述，Parquet是Spark中首选的默认内置数据源文件格式，并且已被许多其他框架采用。我们建议你在ETL和数据提取过程中使用此格式。

3.4  JSON格式

JSON也是一种流行的数据格式。与XML相比，它以易于阅读和易于解析的格式而著称。它具有两种表示形式：单行模式和多行模式。Spark支持两种模式。

在单行模式下，每行表示一个JSON对象，而在多行模式下，整个多行对象构成一个JSON对象。要在此模式下阅读，请把multiLine在option()方法中设置为true 。

将JSON文件读入DATAFRAME

你可以像使用Parquet一样，将JSON文件读入DataFrame中，只需"json"在format()方法中指定即可：

// In Scala
val file = "/databricks-datasets/learning-spark-v2/flights/summary-data/json/*"
val df = spark.read.format("json").load(file)
# In Python
file = "/databricks-datasets/learning-spark-v2/flights/summary-data/json/*"
df = spark.read.format("json").load(file)

将JSON文件读入SPARK SQL表

你也可以像使用Parquet一样从JSON文件创建SQL表：

-- In SQL
CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tblUSING jsonOPTIONS (path  "/databricks-datasets/learning-spark-v2/flights/summary-data/json/*")

创建表后，你可以使用SQL将数据读取到DataFrame中：

// In Scala/Python
spark.sql("SELECT * FROM us_delay_flights_tbl").show()+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |15   |
|United States    |Croatia            |1    |
|United States    |Ireland            |344  |
|Egypt            |United States      |15   |
|United States    |India              |62   |
|United States    |Singapore          |1    |
|United States    |Grenada            |62   |
|Costa Rica       |United States      |588  |
|Senegal          |United States      |40   |
|Moldova          |United States      |1    |
+-----------------+-------------------+-----+
only showing top 10 rows

将DATAFRAMES写入JSON文件

将DataFrame保存为JSON文件很简单。指定适当的DataFrameWriter方法和参数，并提供将JSON文件保存到的位置：

// In Scala
df.write.format("json").mode("overwrite").option("compression", "snappy").save("/tmp/data/json/df_json")
# In Python
(df.write.format("json").mode("overwrite").option("compression", "snappy").save("/tmp/data/json/df_json"))

这将在指定的路径处创建一个目录，该目录中填充了一组紧凑的JSON文件：

-rw-r--r-- 1 jules wheel 0 May 16 14:44 _SUCCESS
-rw-r--r-- 1 jules wheel 71 May 16 14:44 part-00000-<...>-c000.json

JSON数据源选项

表4-3说明了DataFrameReader和DataFrameWriter的常用JSON选项。有关完整列表，请参考文档。

3.5  CSV

与普通文本文件一样，这种通用文本文件格式用逗号分隔的每个数据或字段。每行以逗号分隔的字段代表一条记录。即使逗号是默认的分隔符，如果逗号是数据的一部分，你也可以使用其他定界符来分隔字段。流行的电子表格可以生成CSV文件，因此它是数据和业务分析师中流行的格式。

将CSV文件读入DATAFRAME

与其他内置数据源一样，你可以使用DataFrameReader方法和参数将CSV文件读入DataFrame：

// In Scala
val file = "/databricks-datasets/learning-spark-v2/flights/summary-data/csv/*"
val schema = "DEST_COUNTRY_NAME STRING, ORIGIN_COUNTRY_NAME STRING, count INT"val df = spark.read.format("csv").schema(schema).option("header", "true").option("mode", "FAILFAST")     // Exit if any errors.option("nullValue", "")        // Replace any null data with quotes.load(file)
# In Python
file = "/databricks-datasets/learning-spark-v2/flights/summary-data/csv/*"
schema = "DEST_COUNTRY_NAME STRING, ORIGIN_COUNTRY_NAME STRING, count INT"
df = (spark.read.format("csv").option("header", "true").schema(schema).option("mode", "FAILFAST")  # Exit if any errors.option("nullValue", "")     # Replace any null data field with quotes.load(file))

从CSV数据源创建SQL表与使用Parquet或JSON没什么不同：

-- In SQL
CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tblUSING csvOPTIONS (path "/databricks-datasets/learning-spark-v2/flights/summary-data/csv/*",header "true",inferSchema "true",mode "FAILFAST")

创建表后，你可以像以前一样使用SQL将数据读取到DataFrame中：

// In Scala/Python
spark.sql("SELECT * FROM us_delay_flights_tbl").show(10)+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |1    |
|United States    |Ireland            |264  |
|United States    |India              |69   |
|Egypt            |United States      |24   |
|Equatorial Guinea|United States      |1    |
|United States    |Singapore          |25   |
|United States    |Grenada            |54   |
|Costa Rica       |United States      |477  |
|Senegal          |United States      |29   |
|United States    |Marshall Islands   |44   |
+-----------------+-------------------+-----+
only showing top 10 rows

将DATAFRAMES写入CSV文件

将DataFrame保存为CSV文件很简单。指定适当的DataFrameWriter方法和参数，并提供将CSV文件保存到的位置：

// In Scala
df.write.format("csv").mode("overwrite").save("/tmp/data/csv/df_csv")
# In Python
df.write.format("csv").mode("overwrite").save("/tmp/data/csv/df_csv")

这将在指定位置生成一个文件夹，该文件夹中填充了一堆紧凑的压缩文件：

-rw-r--r-- 1 jules wheel 0 May 16 12:17 _SUCCESS
-rw-r--r-- 1 jules wheel 36 May 16 12:17 part-00000-251690eb-<...>-c000.csv

CSV数据源选项

表4-4介绍了一些常见的DataFrameReader和DataFrameWriter的CSV选项。由于CSV文件可能很复杂，因此可以使用许多选项。有关完整列表，请参考文档。

3.6  Avro

作为内置数据源在Spark 2.4中引入，例如Apache Kafka使用Avro格式用于消息序列化和反序列化。它提供了许多好处，包括直接映射到JSON，提高速度和效率以及绑定许多可用的编程语言。

将AVRO文件读入DATAFRAME

使用Avro文件读取到DataFrame的DataFrameReader用法与我们在本节中讨论的其他数据源的用法是一致的：

// In Scala
val df = spark.read.format("avro").load("/databricks-datasets/learning-spark-v2/flights/summary-data/avro/*")
df.show(false)
# In Python
df = (spark.read.format("avro").load("/databricks-datasets/learning-spark-v2/flights/summary-data/avro/*"))
df.show(truncate=False)+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |1    |
|United States    |Ireland            |264  |
|United States    |India              |69   |
|Egypt            |United States      |24   |
|Equatorial Guinea|United States      |1    |
|United States    |Singapore          |25   |
|United States    |Grenada            |54   |
|Costa Rica       |United States      |477  |
|Senegal          |United States      |29   |
|United States    |Marshall Islands   |44   |
+-----------------+-------------------+-----+
only showing top 10 rows

将AVRO文件读入SPARK SQL表

同样，使用Avro数据源创建SQL表与使用Parquet，JSON或CSV没什么不同：

-- In SQL
CREATE OR REPLACE TEMPORARY VIEW episode_tblUSING avroOPTIONS (path "/databricks-datasets/learning-spark-v2/flights/summary-data/avro/*")

创建表后，可以使用SQL将数据读入DataFrame中：

// In Scala
spark.sql("SELECT * FROM episode_tbl").show(false)
# In Python
spark.sql("SELECT * FROM episode_tbl").show(truncate=False)+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |1    |
|United States    |Ireland            |264  |
|United States    |India              |69   |
|Egypt            |United States      |24   |
|Equatorial Guinea|United States      |1    |
|United States    |Singapore          |25   |
|United States    |Grenada            |54   |
|Costa Rica       |United States      |477  |
|Senegal          |United States      |29   |
|United States    |Marshall Islands   |44   |
+-----------------+-------------------+-----+
only showing top 10 rows

将DATAFRAMES写入AVRO文件

将DataFrame作为Avro文件编写很简单。与往常一样，指定适当的DataFrameWriter方法和参数，并提供将Avro文件保存到的位置：

// In Scala
df.write.format("avro").mode("overwrite").save("/tmp/data/avro/df_avro")
# In Python
(df.write.format("avro").mode("overwrite").save("/tmp/data/avro/df_avro"))

这将在指定位置生成一个文件夹，该文件夹中填充了一堆压缩文件：

-rw-r--r-- 1 jules wheel 0 May 17 11:54 _SUCCESS
-rw-r--r-- 1 jules wheel 526 May 17 11:54 part-00000-ffdf70f4-<...>-c000.avro

AVRO数据源选项

表4-5说明了DataFrameReader和DataFrameWriter的常用选项。文档中提供了完整的选项列表。

3.7  ORC

作为一种额外的优化的列式文件格式，Spark 2.x支持向量化ORC reader（ vectorized

ORC reader）。两种Spark配置决定了要使用哪种ORC实现。当spark.sql.orc.impl设置为native和spark.sql.orc.enableVectorizedReader设置true为时，Spark使用向量化ORC reader。ORC reader一次读取行块（通常每块1024行），而不是一行一行读取，在密集型操作（如扫描，筛选，聚合和联接）中可以简化操作并减少CPU使用率。

对于使用SQL命令USING HIVE OPTIONS (fileFormat 'ORC')创建的Hive ORC SerDe（序列化和反序列化）表，当Spark配置参数spark.sql.hive.convertMetastoreOrc设置为true时，将使用向量化ORC reader。

将ORC文件读入DATAFRAME

要使用ORC向量化reader读取DataFrame，你可以使用常规DataFrameReader方法和选项：

// In Scala
val file = "/databricks-datasets/learning-spark-v2/flights/summary-data/orc/*"
val df = spark.read.format("orc").load(file)
df.show(10, false)
# In Python
file = "/databricks-datasets/learning-spark-v2/flights/summary-data/orc/*"
df = spark.read.format("orc").option("path", file).load()
df.show(10, False)+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |1    |
|United States    |Ireland            |264  |
|United States    |India              |69   |
|Egypt            |United States      |24   |
|Equatorial Guinea|United States      |1    |
|United States    |Singapore          |25   |
|United States    |Grenada            |54   |
|Costa Rica       |United States      |477  |
|Senegal          |United States      |29   |
|United States    |Marshall Islands   |44   |
+-----------------+-------------------+-----+
only showing top 10 rows

将ORC文件读取到SPARK SQL表中

使用ORC数据源创建SQL视图时，与Parquet，JSON，CSV或Avro没有什么区别：

-- In SQL
CREATE OR REPLACE TEMPORARY VIEW us_delay_flights_tblUSING orcOPTIONS (path "/databricks-datasets/learning-spark-v2/flights/summary-data/orc/*")

创建表后，你可以照常使用SQL将数据读取到DataFrame中：

// In Scala/Python
spark.sql("SELECT * FROM us_delay_flights_tbl").show()+-----------------+-------------------+-----+
|DEST_COUNTRY_NAME|ORIGIN_COUNTRY_NAME|count|
+-----------------+-------------------+-----+
|United States    |Romania            |1    |
|United States    |Ireland            |264  |
|United States    |India              |69   |
|Egypt            |United States      |24   |
|Equatorial Guinea|United States      |1    |
|United States    |Singapore          |25   |
|United States    |Grenada            |54   |
|Costa Rica       |United States      |477  |
|Senegal          |United States      |29   |
|United States    |Marshall Islands   |44   |
+-----------------+-------------------+-----+
only showing top 10 rows

将DATAFRAME写入ORC文件

使用以下DataFrameWriter方法，读取后写回转换后的DataFrame也同样简单：

// In Scala
df.write.format("orc").mode("overwrite").option("compression", "snappy").save("/tmp/data/orc/df_orc")
# In Python
(df.write.format("orc").mode("overwrite").option("compression", "snappy").save("/tmp/data/orc/flights_orc"))

结果将是指定位置的文件夹，其中包含一些压缩的ORC文件：

-rw-r--r-- 1 jules wheel 0 May 16 17:23 _SUCCESS
-rw-r--r-- 1 jules wheel 547 May 16 17:23 part-00000-<...>-c000.snappy.orc

3.8  Images

在Spark 2.4中，社区引入了一个新的数据源图像文件，以支持深度学习和机器学习框架，例如TensorFlow和PyTorch。对于基于计算机视觉的机器学习应用程序，加载和处理图像数据集非常重要。

将图像文件读入DATAFRAME

与所有以前的文件格式一样，你可以使用DataFrameReader方法和选项来读取图像文件，如下所示：

// In Scala
import org.apache.spark.ml.source.imageval imageDir = "/databricks-datasets/learning-spark-v2/cctvVideos/train_images/"
val imagesDF = spark.read.format("image").load(imageDir)imagesDF.printSchemaimagesDF.select("image.height", "image.width", "image.nChannels", "image.mode", "label").show(5, false)# In Python
from pyspark.ml import imageimage_dir = "/databricks-datasets/learning-spark-v2/cctvVideos/train_images/"
images_df = spark.read.format("image").load(image_dir)
images_df.printSchema()root|-- image: struct (nullable = true)|    |-- origin: string (nullable = true)|    |-- height: integer (nullable = true)|    |-- width: integer (nullable = true)|    |-- nChannels: integer (nullable = true)|    |-- mode: integer (nullable = true)|    |-- data: binary (nullable = true)|-- label: integer (nullable = true)images_df.select("image.height", "image.width", "image.nChannels", "image.mode", "label").show(5, truncate=False)+------+-----+---------+----+-----+
|height|width|nChannels|mode|label|
+------+-----+---------+----+-----+
|288   |384  |3        |16  |0    |
|288   |384  |3        |16  |1    |
|288   |384  |3        |16  |0    |
|288   |384  |3        |16  |0    |
|288   |384  |3        |16  |0    |
+------+-----+---------+----+-----+
only showing top 5 rows

3.9  二进制文件

Spark 3.0添加了对二进制文件作为数据源的支持。DataFrameReader将二进制文件转换为包含该文件的原始内容和元数据的单个DataFrame Row（记录）。二进制文件源生成的DataFrame包含如下字段：

• path：StringType

• modificationTime：TimestampType

• length：LongType

• content：BinaryType

将二进制文件读入DATAFRAME

要读取二进制文件，请将数据源格式指定为binaryFile。你可以使用与给定全局模式匹配的路径加载文件，同时保留分区信息，使用数据源选项pathGlobFilter去匹配。例如，以下代码从输入目录中读取所有带有分区目录的JPG文件：

// In Scala
val path = "/databricks-datasets/learning-spark-v2/cctvVideos/train_images/"
val binaryFilesDF = spark.read.format("binaryFile").option("pathGlobFilter", "*.jpg").load(path)
binaryFilesDF.show(5)
# In Python
path = "/databricks-datasets/learning-spark-v2/cctvVideos/train_images/"
binary_files_df = (spark.read.format("binaryFile").option("pathGlobFilter", "*.jpg").load(path))
binary_files_df.show(5)+--------------------+-------------------+------+--------------------
| path|   modificationTime|length|             content|label|
+--------------------+-------------------+------+--------------------+-----+
|file:/Users/jules...|2020-02-12 12:04:24| 55037|[FF D8 FF E0 00 1...|    0|
|file:/Users/jules...|2020-02-12 12:04:24| 54634|[FF D8 FF E0 00 1...|    1|
|file:/Users/jules...|2020-02-12 12:04:24| 54624|[FF D8 FF E0 00 1...|    0|
|file:/Users/jules...|2020-02-12 12:04:24| 54505|[FF D8 FF E0 00 1...|    0|
|file:/Users/jules...|2020-02-12 12:04:24| 54475|[FF D8 FF E0 00 1...|    0|
+--------------------+-------------------+------+--------------------+-----+
only showing top 5 rows

要忽略目录中的分区数据发现，可以设置recursiveFileLookup为"true"：

// In Scala
val binaryFilesDF = spark.read.format("binaryFile").option("pathGlobFilter", "*.jpg").option("recursiveFileLookup", "true").load(path)
binaryFilesDF.show(5)
# In Python
binary_files_df = (spark.read.format("binaryFile").option("pathGlobFilter", "*.jpg").option("recursiveFileLookup", "true").load(path))
binary_files_df.show(5)+--------------------+-------------------+------+--------------------+
|                path|   modificationTime|length|             content|
+--------------------+-------------------+------+--------------------+
|file:/Users/jules...|2020-02-12 12:04:24| 55037|[FF D8 FF E0 00 1...|
|file:/Users/jules...|2020-02-12 12:04:24| 54634|[FF D8 FF E0 00 1...|
|file:/Users/jules...|2020-02-12 12:04:24| 54624|[FF D8 FF E0 00 1...|
|file:/Users/jules...|2020-02-12 12:04:24| 54505|[FF D8 FF E0 00 1...|
|file:/Users/jules...|2020-02-12 12:04:24| 54475|[FF D8 FF E0 00 1...|
+--------------------+-------------------+------+--------------------+
only showing top 5 rows

请注意，当recursiveFileLookup选项设置为"true"时label列不存在。

当前，二进制文件数据源不支持将DataFrame写回原始文件格式。

在本节中，你将了解如何从一系列受支持的文件格式将数据读取到DataFrame中。我们还向你展示了如何从现有内置数据源创建临时视图和表。无论你使用的是DataFrame API还是SQL，查询都会产生相同的结果。你可以在本书的GitHub存储库中的笔记中检查其中一些查询。

4.  总结

回顾一下，本章探讨了DataFrame API和Spark SQL之间的互操作性。特别是，你了解了如何使用Spark SQL进行以下操作：

• 使用Spark SQL和DataFrame API创建托管表和非托管表。

• 读取和写入各种内置数据源和文件格式。

• 使用spark.sql编程接口对存储为Spark SQL表或视图的结构化数据执行SQL查询。

• 细读Spark Catalog并探索与表和视图关联的元数据。

• 使用DataFrameWriter和DataFrameReader API。

通过本章中的代码片段以及该书的GitHub仓库中的笔记，你可以了解如何使用DataFrames和Spark SQL。继续这一思路，下一章将进一步探讨Spark如何与图4-1中所示的外部数据源进行交互。你将看到一些更深入的转换示例以及DataFrame API和Spark SQL之间的互操作性。