本章正式介绍Spark可以使用的开箱即用的各种其他数据源，以及由更大的社区构建的无数其他数据源。Spark有6个“核心”数据源和数百个由社区编写的外部数据源。能够从所有不同类型的数据源进行读写，这可以说是Spark最大的优势之一。以下是Spark的核心数据源:
CSV
JSON
Parquet
ORC
JDBC/ODBC connections
Plain-text files
如前所述，Spark有许多社区创建的数据源。这里只是一个小例子:
Cassandra
HBase
MongoDB
AWS Redshift
XML
还有很多很多其他的
本章的目标是让您能够从Spark的核心数据源读写数据，并了解在与第三方数据源集成时应该寻找的内容。为了实现这一点，我们将重点放在您需要能够识别和理解的核心概念上。
9.1. 数据源API的结构
在继续学习如何从某些格式读写之前，让我们先看看数据源api的总体组织结构。
9.1.1. Read API Structure
读取数据的核心结构如下:
DataFrameReader.format(...).option("key", "value").schema(...).load()
我们将使用这种格式从所有数据源读取数据。format是可选的，因为默认情况下Spark将使用Parquet格式。option允许您设置键值配置，以参数化读取数据的方式。最后，如果数据源提供模式，或者您打算使用模式推断，那么schema是可选的。当然，每种格式都有一些必需的选项，我们将在查看每种格式时讨论这些选项。
提示
Spark社区中有很多简写符号，数据源read API也不例外。我们试着在整本书中保持一致，同时仍然揭示了一些简写符号。
9.1.2. 读数据的基础
在Spark中读取数据的基础是DataFrameReader。我们通过read属性通过SparkSession访问它:
spark.read
有了DataFrame读取器后，我们指定几个值:
The format格式
The schema 模式
The read mode
一系列配置选项
格式、选项和模式每个返回一个DataFrameReader，该ader可以进行进一步的转换，并且都是可选的，只有一个选项除外。每个数据源都有一组特定的选项，它们决定如何将数据读入Spark(我们将很快介绍这些选项)。至少，您必须为DataFrameReader提供一条要从中读取的路径。下面是一个整体样式的例子:
spark.read.format("csv") .option("mode", "FAILFAST") .option("inferSchema", "true") .option("path", "path/to/file(s)") .schema(someSchema) .load()
设置选项的方法有很多种;例如，您可以构建一个map并传递您的配置。现在，我们将坚持您刚才看到的简单而明确的方法。读取模式从外部数据源读取数据自然会遇到格式不正确的数据，特别是在只使用半结构化数据源时。读取模式指定当Spark遇到格式不正确的记录时会发生什么。表9-1列出了读取模式。
读取模式
描述
Permissive(默认)
当遇到损坏的记录时，将所有字段设置为null，并将所有损坏的记录放在名为_corruption t_record的字符串列中
dropMalformed
删除包含损坏记录的行
failFast
遇到不正确的记录时立即失败
9.1.3. Write API Structure
写数据的核心结构如下:
DataFrameWriter.format(.).option(.).partitionBy(.).bucketBy(.).sortBy(.).save()
我们将使用这种格式写入所有数据源。format是可选的，因为默认情况下Spark将使用Parquet格式。option允许我们配置如何写出给定的数据。仅对基于文件的数据源使用PartitionBy、bucketBy和sortBy;您可以使用它们来控制目标文件的特定布局。
9.1.4.写数据的基础
写数据的基础与读数据的基础非常相似。我们使用的是DataFrameWriter，而不是DataFrameReader。因为我们总是需要写出一些给定的数据源，所以我们通过write属性在每个dataframe的基础上访问DataFrameWriter:
//in Scala dataFrame.write
有了DataFrameWriter之后，我们指定三个值: 格式、一系列选项和保存模式。至少，您必须提供一条路径。我们将很快讨论选项的可能性，这些选项因数据源而异。
// in Scaladataframe.write.format("csv") .option("mode", "OVERWRITE") .option("dateFormat", "yyyy-MM-dd") .option("path", "path/to/file(s)") .save()
Save modes
保存模式指定如果Spark在指定位置找到数据会发生什么(假设其他条件都相同)。表9-2列出了保存模式。
Save mode
描述
append
将输出文件附加到该位置已经存在的文件列表中
overwrite
会完全覆盖任何已经存在的数据
errorIfExists
如果数据或文件已经存在于指定位置，则引发错误并无法写入
ignore
如果该位置存在数据或文件，则当前DataFrame不做任何操作
默认值是errorIfExists。这意味着，如果Spark在您要写入的位置找到数据，它将立即导致写入失败。我们已经大致介绍了在使用数据源时所需的核心概念，现在让我们深入研究每个Spark内置支持的数据源。
9.2. CSV 文件
CSV表示逗号分隔的值。这是一种常见的文本文件格式，其中每一行表示一条记录，记录中的每个字段用逗号分隔。CSV文件虽然看起来结构很好，但实际上是您将遇到的最棘手的文件格式之一，因为在生产场景中无法对它们包含什么或如何进行结构假设。因此，CSV reader有很多选项。这些选项使您能够解决一些问题，比如某些字符需要转义(例如，当文件也是逗号分隔的或以非常规方式标记的空值时，列内的逗号)。
9.2.1. CSV选项
表9-3给出了CSV reader中可用的选项。
读/写
Key
可能值
默认值
描述
Both
sep
任何单个字符串字符
,
用于分隔每个字段和值的单个字符。
Both
header
true,false
false
一个布尔标志，它声明文件中的第一行是否为列的名称。
Read
escape
任何字符
\
转义字符
Read
inferSchema
true, false
false
指定Spark在读取文件时是否应该推断列类型。
Read
ignoreLeadingWhiteSpace
true, false
false
声明是否跳过正在读取的值的前导空格。
Read
ignoreTrailingWhiteSpace
true, false
false
声明是否应跳过正在读取的值的尾随空格。
Both
nullValue
任何字符
“”
声明文件中哪个字符表示null。
Both
nanValue
任何字符
NAN
声明CSV文件中表示NaN或缺失字符的字符。
Both
positiveInf
任何字符串或字符
Inf
声明哪个字符表示正无穷值
Both
negativeInf
任何字符串或字符
-Inf
声明哪个字符表示负无穷值
Both
compression or codec
None, uncompressed, bzip2, deflate, gzip, lz4, or snappy
None
声明Spark应使用什么压缩编解码器来读取或写入文件。
Both
dateFormat
任何符合java SimpleDataFormat的字符串或字符。
yyyy-MM-dd
为任何属于日期类型的列声明日期格式。
Both
timestampFormat
任何符合java SimpleDataFormat的字符串或字符。
yyyy-MM-dd’T’HH:mm:ss.SSSZZ
为任何属于时间戳类型的列声明时间戳格式。
Read
maxColumns
任意整数
20480
声明文件中的最大列数
Read
maxCharsPerColumn
任意整数
1000000
声明列中的最大字符数
Read
escapeQuotes
true, false
false
声明Spark是否应该转义行中找到的引号。
Read
maxMalformedLogPerPartition
任意整数
10
设置每个分区将记录的最大畸形行数。超过这个数字的畸形记录将被忽略。
Write
quoteAll
true, false
false
指定是否应该将所有值都括在引号中，而不只是转义具有引号字符的值。
Read
multiLine
true, false
false
此选项允许您读取多行CSV文件，其中CSV文件中的每个逻辑行可能跨越文件本身的多行。
9.2.2.读CSV文件
要像读取其他格式一样读取CSV文件，我们必须首先为该特定格式创建一个DataFrameReader。这里，我们指定格式为CSV:
spark.read.format("csv")
在此之后，我们可以指定模式和选项。让我们设置几个选项，一些是我们在书的开头看到的，另一些是我们还没有看到的。我们将CSV文件的头部设置为true，模式为FAILFAST，模式推断为true:
// in Scalaspark.read.format("csv") .option("header", "true") .option("mode", "FAILFAST") .option("inferSchema", "true") .load("some/path/to/file.csv")
如前所述，我们可以使用该模式来指定对畸形数据的容忍度。例如，我们可以使用这些模式和我们在第5章中创建的模式来确保我们的文件符合我们期望的数据:
// in Scalaimport org.apache.spark.sql.types.{StructField, StructType, StringType, LongType}val myManualSchema = new StructType(Array( new StructField("DEST_COUNTRY_NAME", StringType, true), new StructField("ORIGIN_COUNTRY_NAME", StringType, true), new StructField("count", LongType, false)))spark.read.format("csv") .option("header", "true") .option("mode", "FAILFAST") .schema(myManualSchema) .load("/data/flight-data/csv/2010-summary.csv") .show(5)
当我们不期望数据以某种格式出现时，事情就变得棘手了，但无论如何，它是以这种方式出现的。例如，让我们使用当前模式并将所有列类型更改为LongType。这与实际的模式不匹配，但是Spark对我们这样做没有问题。只有当Spark实际读取数据时，这个问题才会显现出来。一旦我们启动Spark作业，由于数据不符合指定的模式，它将立即失败(在我们执行作业之后):
// in Scalaval myManualSchema = new StructType(Array( new StructField("DEST_COUNTRY_NAME", LongType, true), new StructField("ORIGIN_COUNTRY_NAME", LongType, true), new StructField("count", LongType, false) ))spark.read.format("csv") .option("header", "true") .option("mode", "FAILFAST") .schema(myManualSchema) .load("/data/flight-data/csv/2010-summary.csv") .take(5)
通常，Spark只会在作业执行时失败，而不是在DataFrame定义时失败——即使我们指向一个不存在的文件。这是由于延迟计算，我们在第2章中学习了这个概念。
9.2.3.写CSV文件
就像读取数据一样，当我们写入CSV文件时，有许多选项(表9-3中列出)可以用于写入数据。这是读取选项的子集，因为许多选项在写入数据时不适用(如maxColumns和inferSchema)。这里有一个例子:
// in Scalaval csvFile = spark.read.format("csv") .option("header", "true").option("mode", "FAILFAST").schema(myManualSchema) .load("/data/flight-data/csv/2010-summary.csv")# in PythoncsvFile = spark.read.format("csv")\ .option("header", "true")\ .option("mode", "FAILFAST")\ .option("inferSchema", "true")\ .load("/data/flight-data/csv/2010-summary.csv")
例如，我们可以把我们的CSV文件写出去，作为一个TSV文件，这很容易:
// in ScalacsvFile.write.format("csv").mode("overwrite").option("sep", "\t") .save("/tmp/my-tsv-file.tsv")# in PythoncsvFile.write.format("csv").mode("overwrite").option("sep", "\t")\ .save("/tmp/my-tsv-file.tsv")
当您列出目标目录时，您可以看到my-tsv-file实际上是一个包含许多文件的文件夹:
$ ls /tmp/my-tsv-file.tsv//tmp/my-tsv-file.tsv/part-00000-35cf9453-1943-4a8c-9c82-9f6ea9742b29.csv
这实际上反映了在我们创建DataFrame时，DataFrame中分区的数量。如果在此之前对数据进行重新分区，最终会得到不同数量的文件。我们将在本章末尾讨论这种取舍。
9.3. JSON文件
来自JavaScript领域的人可能熟悉JavaScript对象表示法，也就是通常所说的JSON。在处理这类数据时，有一些问题值得我们在开始之前考虑。在Spark中，当我们引用JSON文件时，我们引用以行分隔的JSON文件。这与每个文件都有一个大型JSON对象或JSON数组的文件形成了对比。行分隔与多行权衡由一个选项控制：multiLine。当您将该选项设置为true时，您可以将整个文件作为一个json对象读取，Spark将完成将其解析为一个DataFrame的工作。以行分隔的JSON实际上是一种更稳定的格式，因为它允许您向文件附加一条新记录(而不是必须读取整个文件然后将其写出来)，这是我们推荐您使用的格式。行分隔JSON流行的另一个关键原因是JSON对象具有结构，而JavaScript (JSON基于JavaScript)至少具有基本类型。这使得处理起来更容易，因为Spark可以代表我们对数据做更多的假设。您会注意到，由于对象的关系，JSON选项要比CSV的选项少得多。
9.3.1.JSON配置选项
表9-4列出了JSON对象可用的选项及其描述。
读/写
Key
可能值
默认值
描述
Both
compression 或 codec
None, uncompressed, bzip2, deflate, gzip, lz4, or snappy
None
声明Spark应使用什么压缩编解码器来读取或写入文件。
Both
dateFormat
任何符合java SimpleDataFormat的字符串或字符。
yyyy-MM-dd
为任何属于日期类型的列声明日期格式。
Both
timestampFormat
任何符合java SimpleDataFormat的字符串或字符。
yyyy-MM-dd’T’HH:mm:ss.SSSZZ
为任何属于时间戳类型的列声明时间戳格式。
Read
primitiveAsString
true, false
false
将所有基本值推断为字符串类型。
Read
allowComments
true, false
false
忽略JSON记录中的Java/ c++样式注释。
Read
allowUnquotedFieldNames
true, false
false
允许非引号JSON字段名
Read
allowSingleQuotes
true, false
true
除双引号外，还允许单引号。
Read
allowNumericLeadingZeros
true, false
false
允许数字的前导零(例如，00012)。
Read
allowBackslashEscapingAnyCharacter
true, false
false
允许使用反斜杠引用机制接受所有字符的引用。
Read
columnNameOfCorruptRecord
任意字符串
spark.sql
.column&NameOfCorruptRecord值
允许使用permissive模式创建的畸形字符串重命名新字段。这将覆盖配置值。
Read
multiLine
true, false
false
允许读取非行分隔的JSON文件。
现在，读取以行分隔的JSON文件只在格式和我们指定的选项上有所不同:
spark.read.format("json")
9.3.2.读取 JSON文件
让我们看一个读取JSON文件的例子，并比较我们看到的配置选项:
// in Scalaspark.read.format("json").option("mode", "FAILFAST").schema(myManualSchema) .load("/data/flight-data/json/2010-summary.json").show(5)# in Pythonspark.read.format("json").option("mode", "FAILFAST")\ .option("inferSchema", "true")\ .load("/data/flight-data/json/2010-summary.json").show(5)
9.3.3.写 JSON文件
编写JSON文件与读取它们一样简单，而且，正如您所期望的那样，数据源并不重要。因此，我们可以重用前面创建的CSV DataFrame作为JSON文件的源。这也遵循我们之前指定的规则:每个分区写一个文件，整个DataFrame作为一个文件夹写。每一行也有一个JSON对象:
// in ScalacsvFile.write.format("json").mode("overwrite").save("/tmp/my-json-file.json")# in PythoncsvFile.write.format("json").mode("overwrite").save("/tmp/my-json-file.json")$ ls /tmp/my-json-file.json//tmp/my-json-file.json/part-00000-tid-543....json
9.4. Parquet文件
Parquet是一个开源的面向列的数据存储，它提供了多种存储优化，特别是对于分析工作负载。它提供了columnar压缩，节省了存储空间，允许读取单独的列而不是整个文件。它是一种与Apache Spark非常配合的文件格式，实际上是默认的文件格式。我们建议将数据写入Parquet进行长期存储，因为从Parquet文件读取总是比JSON或CSV更有效。Parquet的另一个优点是它支持复杂的类型。这意味着，如果您的列是一个数组(例如，CSV文件可能会失败)、map或struct，您仍然能够毫无问题地读写该文件。下面是如何指定Parquet作为读取格式:
spark.read.format("parquet")
9.4.1. 读取parquet文件
Parquet几乎没有什么选项，因为它在存储数据时强制执行自己的模式。因此，您只需要设置格式就可以了。如果我们对DataFframe的外观有严格的要求，我们可以设置模式。通常这是不必要的，因为我们可以在读取时使用模式，这与使用CSV文件的推断模式类似。但是，对于Parquet文件，这种方法更强大，因为模式构建到文件本身(因此不需要推理)。下面是一些读取Parquet文件的简单例子:
spark.read.format("parquet")// in Scalaspark.read.format("parquet") .load("/data/flight-data/parquet/2010-summary.parquet").show(5)# in Pythonspark.read.format("parquet")\ .load("/data/flight-data/parquet/2010-summary.parquet").show(5)
9.4.2. Parquet配置项
正如我们刚才提到的，很少有parquet的配置选项—确切地说，只有两个—因为它有一个定义良好的规范，与Spark中的概念紧密一致。表9-5给出了这些选项。
警告
尽管只有两种选择，但如果使用不兼容的parquet文件，仍然会遇到问题。当您使用不同版本的Spark(尤其是旧版本)写parquet文件时要小心，因为这可能会引起严重的麻烦。
读/写
Key
可能值
默认值
描述
Write
compression
或 codec
None, uncompressed, bzip2, deflate, gzip, lz4, or snappy
None
声明Spark应使用什么压缩编解码器来读取或写入文件。
Read
mergeSchema
true, false
配置spark.sql.parquet.mergeSchema的值
您可以在相同的表/文件夹中增量地将列添加到新编写的parquet文件中。使用此选项可启用或禁用此功能。
9.4.3.写parquet文件
Parquet文件写起来和读起来一样容易。我们只需指定文件的位置。适用相同的分区规则:
// in ScalacsvFile.write.format("parquet").mode("overwrite") .save("/tmp/my-parquet-file.parquet")# in PythoncsvFile.write.format("parquet").mode("overwrite")\ .save("/tmp/my-parquet-file.parquet")
9.5. ORC文件
ORC是一种为Hadoop工作负载设计的自描述、类型感知的列文件格式。它针对大型流读取进行了优化，但是集成了快速查找所需行的支持。ORC实际上没有读取数据的选项，因为Spark非常理解文件格式。一个经常被问到的问题是:ORC和Parquet有什么不同?在很大程度上，它们非常相似;基本的区别是，Parquet是使用Spark的进一步优化，而ORC是使用Hive的进一步优化。
9.5.1. 读取ORC文件
下面是如何将ORC文件读入Spark:
// in Scalaspark.read.format("orc").load("/data/flight-data/orc/2010-summary.orc").show(5)# in Pythonspark.read.format("orc").load("/data/flight-data/orc/2010-summary.orc").show(5)
9.5.2. 写ORC文件
在本章的这一点上，您应该很容易猜到如何写ORC文件。它确实遵循了我们目前看到的完全相同的模式，即我们指定格式，然后保存文件:
// in ScalacsvFile.write.format("orc").mode("overwrite").save("/tmp/my-json-file.orc")# in PythoncsvFile.write.format("orc").mode("overwrite").save("/tmp/my-json-file.orc")
9.6. SQL数据库
SQL数据源是功能更强大的连接器之一，因为可以连接到多种系统(只要该系统使用SQL)。例如，您可以连接到MySQL数据库、PostgreSQL数据库或Oracle数据库。您还可以连接到SQLite，在本例中我们将这样做。当然，数据库不仅仅是一组原始文件，所以对于如何连接到数据库，有更多的选项需要考虑。也就是说，您需要开始考虑诸如身份验证和连接之类的问题(您需要确定Spark集群的网络是否连接到数据库系统的网络)。为了避免为本书的目的而设置数据库，我们提供了一个在SQLite上运行的参考示例。通过使用SQLite，我们可以跳过很多这样的细节，因为它可以在本地机器上以最小的设置工作，同时限制了不能在分布式设置中工作。如果希望在分布式环境中处理这些示例，则需要连接到另一种数据库。
SQLITE简介
SQLite是全世界使用最多的数据库引擎，这是有原因的。它功能强大、速度快、易于理解。这是因为SQLite数据库只是一个文件。这将使您非常容易地启动和运行，因为我们在本书的repository中包含了源文件。只需将该文件下载到本地计算机，就可以从该文件中读取和写入。我们使用SQLite，但是这里的所有代码都与更传统的关系数据库(如MySQL)一起工作。主要区别在于连接到数据库时包含的属性。当我们使用SQLite时，没有用户或密码的概念。
警告
虽然SQLite是一个很好的参考示例，但它可能不是您想在生产中使用的。此外，SQLite在分布式设置中不一定能很好地工作，因为它需要在写时锁定整个数据库。我们在这里展示的示例也将以类似的方式使用MySQL或PostgreSQL。
要从这些数据库读写，您需要做两件事:在spark类路径上为您的特定数据库包含Java数据库连接(JDBC)驱动程序，并为驱动程序本身提供适当的JAR。例如，为了能够从PostgreSQL读取和写入，您可以运行如下代码:
./bin/spark-shell \--driver-class-path postgresql-9.4.1207.jar \--jars postgresql-9.4.1207.jar
与我们的其他源一样，在从SQL数据库读取和写入时，有许多选项可用。其中只有一些与我们当前的示例相关，但是表9-6列出了在使用JDBC数据库时可以设置的所有选项。
属性名称
属性含义
url
要连接到的JDBC URL。特定于源的连接属性可以在URL中指定;例如：jdbc:postgresql://localhost/test?user=fred&password=secret.
dbtable
要读取的JDBC表。注意，可以使用SQL查询的FROM子句中有效的任何内容。例如，您还可以在括号中使用子查询，而不是完整的表。
driver
用于连接到此URL的JDBC驱动程序的类名。
partitionColumn, lowerBound, upperBound
如果指定了其中一个选项，那么还必须设置所有其他选项。此外，必须指定numpartition。这些属性描述了如何在从多个worker并行读取数据时对表进行分区。partitionColumn必须是表中的数字列。注意，lowerBound和lowerBound仅用于确定分区步长，而不是用于过滤表中的行。因此，表中的所有行都将被分区并返回。此选项仅适用于read。
numPartitions
可用于表读写并行性的最大分区数。这也决定了并发JDBC连接的最大数量。
fetchsize
此配置项决定每次往返要获取多少行。这可以提高JDBC驱动程序的性能，JDBC驱动程序默认取值大小较低(例如，Oracle有10行)。此选项仅适用于read。
batchsize
JDBC批处理大小，它决定每次往返插入多少行。这有助于JDBC驱动程序的性能。这个选项只适用于write。默认值是1000。
isolationLevel
事务隔离级别，适用于当前连接。它可以是NONE、READ_COMMITTED、READ_UNCOMMITTED、REPEATABLE_READ或SERIALIZABLE中的一个，对应于JDBC连接对象定义的标准事务隔离级别。默认值是READ_UNCOMMITTED。这个选项只适用于Write。有关更多信息，请参阅java.sql.Connection中的文档。
truncate
这是一个与JDBC writer相关的选项。当启用SaveMode.Overwrite，Spark将truncate现有表，而不是删除和重新创建它。这可以更有效，并且可以防止删除表元数据(例如索引)。但是，在某些情况下，例如当新数据具有不同的schema时，它将不起作用。默认值为false。这个选项只适用于write。
createTableOptions
这是一个与JDBC writer相关的选项。如果指定，这个选项允许在创建表时设置特定于数据库的表和分区选项(例如，CREATE table t (name string) ENGINE=InnoDB)。这个选项只适用于write。
createTableColumnTypes
创建表时要使用的数据库列数据类型，替换默认数据类型。数据类型信息应该以与CREATE TABLE columns语法相同的格式指定(例如，“name CHAR(64)， comments VARCHAR(1024)”)。指定的类型应该是有效的Spark SQL数据类型。这个选项只适用于write。
9.6.1.从SQL数据库中读取
在读取文件时，SQL数据库与我们前面看到的其他数据源没有什么不同。对于这些源，我们指定格式和选项，然后加载数据:
// in Scalaval driver = "org.sqlite.JDBC"val path = "/data/flight-data/jdbc/my-sqlite.db"val url = s"jdbc:sqlite:/${path}"val tablename = "flight_info"# in Pythondriver = "org.sqlite.JDBC"path = "/data/flight-data/jdbc/my-sqlite.db"url = "jdbc:sqlite:" + pathtablename = "flight_info"
在定义连接属性之后，可以测试到数据库本身的连接，以确保其功能正常。这是一种优秀的故障排除技术，可以确认您的数据库(至少)对Spark driver程序可用。这与SQLite关系不大，因为SQLite是您机器上的一个文件，但是如果您使用的是MySQL之类的工具，您可以使用以下工具测试连接:
import java.sql.DriverManagerval connection = DriverManager.getConnection(url)connection.isClosed()connection.close()
如果连接成功，您就可以开始了。让我们从SQL表中读取DataFrame:
// in Scalaval dbDataFrame = spark.read.format("jdbc").option("url", url) .option("dbtable", tablename).option("driver", driver).load()# in PythondbDataFrame = spark.read.format("jdbc").option("url", url)\ .option("dbtable", tablename).option("driver", driver).load()
SQLite的配置相当简单(例如，没有用户)。其他数据库，比如PostgreSQL，需要更多的配置参数。让我们执行刚才执行的读取操作，这次使用PostgreSQL:
// in Scalaval pgDF = spark.read .format("jdbc") .option("driver", "org.postgresql.Driver") .option("url", "jdbc:postgresql://database_server") .option("dbtable", "schema.tablename") .option("user", "username").option("password","my-secret-password").load()# in PythonpgDF = spark.read.format("jdbc")\ .option("driver", "org.postgresql.Driver")\ .option("url", "jdbc:postgresql://database_server")\ .option("dbtable", "schema.tablename")\ .option("user", "username").option("password", "my-secret-password").load()
当我们创建这个DataFrame时，它与其他数据没有什么不同:您可以对其执行查询操作、transform操作、join操作。您还会注意到，这里已经有了一个模式。这是因为Spark从表本身收集这些信息，并将类型映射到Spark数据类型。让我们只得到不同的location字段，以验证我们可以查询它:
太棒了，我们可以查询数据库!在我们继续之前，有一些细微的细节是值得理解的。
9.6.2.Query Pushdown
首先，Spark在创建DataFrame之前尽最大努力过滤数据库本身中的数据。例如，在之前的查询示例中，我们可以从查询计划中看到，它只从表中选择了相关的列名:
Spark实际上可以在某些查询上做得更好。例如，如果我们在DataFrame上指定一个过滤器，Spark将把该过滤器下推到数据库中。我们可以在PushedFilters下的explain计划中看到这一点。
Spark不能将它自己的所有函数转换为您正在使用的SQL数据库中可用的函数。因此，有时需要将整个查询传递到SQL中，以DataFrame的形式返回结果。这看起来有点复杂，但实际上很简单。您只需指定一个SQL查询，而不是指定一个表名。当然，您需要以一种特殊的方式指定它;你必须把查询用括号括起来，然后把它重命名——在这种情况下，我只是给了它相同的表名:
现在，当您查询这个表时，实际上是在查询该查询的结果。我们可以在解释计划中看到这一点。Spark甚至不知道表的实际模式，只知道我们之前查询的结果:
9.6.3.并行读取数据库
在本书中，我们一直在讨论分区及其在数据处理中的重要性。Spark有一个底层算法，可以将多个文件读入一个分区，或者反过来，根据文件大小和文件类型和压缩的“可分割性”从一个文件读入多个分区。与文件具有相同的灵活性，SQL数据库也具有相同的灵活性，只是您必须手动配置它。正如前面的选项所示，您可以配置的是指定最大分区数的能力，以允许您限制并行读写的数量:
在这种情况下，这个分区仍然作为一个分区，因为没有太多的数据。但是，这种配置可以帮助您确保在读取和写入数据时不会压垮数据库:
dbDataFrame.select("DEST_COUNTRY_NAME").distinct().show()
不幸的是，还有其他一些优化似乎只在另一个API集中进行。您可以通过连接本身显式地将谓词下推到SQL数据库中。这种优化允许您通过指定谓词来控制特定分区中特定数据的物理位置。这有点拗口，我们来看一个简单的例子。我们的数据中只需要两个国家的数据:安圭拉和瑞典。我们可以过滤它们并将它们推入数据库，但是我们还可以更进一步，让它们到达Spark中自己的分区。我们通过在创建数据源时指定谓词列表来实现:
如果指定的谓词不是不相交的，则可能会出现大量重复的行。下面是一个谓词示例集，它将导致重复的行:
9.6.4.基于滑动窗口的分区
让我们看看如何基于谓词进行分区。在本例中，我们将根据数值类型的count列进行分区。这里，我们为第一个分区和最后一个分区都指定了最小值和最大值。任何超出这些界限的都在第一个分区或最后一个分区中。然后，我们设置希望的分区总数(这是并行度的级别)。然后Spark并行查询我们的数据库并返回numpartition分区。我们只需修改上界和下界，以便在某些分区中放置某些值。没有像我们在前面的例子中看到的那样进行过滤:
这将把区间从低到高平均分配:
// in Scalaspark.read.jdbc(url,tablename,colName,lowerBound,upperBound,numPartitions,props) .count() // 255# in Pythonspark.read.jdbc(url, tablename, column=colName, properties=props, lowerBound=lowerBound, upperBound=upperBound, numPartitions=numPartitions).count() # 255
9.6.5.写入SQL数据库
写SQL数据库和以前一样简单。您只需指定URI并根据所需的指定写入模式写出数据。在下面的示例中，我们指定overwrite，它覆盖整个表。我们将使用之前定义的CSV DataFframe来实现这一点:
让我们看看结果:
当然，我们也可以很容易地将这个新表记录追加到表中:
9.7. Text文件
Spark还允许您读取纯文本文件。文件中的每一行都成为DataFrame中的一条记录。然后就由您来相应地转换它。作为实现此目的的一个示例，假设您需要将一些Apache日志文件解析为某种更结构化的格式，或者您可能希望解析一些纯文本以进行自然语言处理。文本文件为Dataset API提供了很好的参数，因为它能够利用native类型的灵活性。
9.7.1.读文本文件
读取文本文件很简单:只需指定textFile的类型。对于textFile，分区目录名将被忽略。要根据分区读写文本文件，应该使用text，其在读写上关注分区:
spark.read.textFile("/data/flight-data/csv/2010-summary.csv") .selectExpr("split(value, ',') as rows").show()
9.7.2. 写文本文件
编写文本文件时，需要确保只有一个字符串列;否则，写入将失败:
csvFile.select("DEST_COUNTRY_NAME").write.text("/tmp/simple-text-file.txt")
如果在执行写操作时执行一些分区(我们将在接下来的几页中讨论分区)，则可以写更多的列。然而，这些列将显示为您要写入的文件夹中的目录，而不是每个文件上的列:
9.8. 高级IO概念
我们在前面已经看到，我们可以通过在写之前控制分区来控制所写文件的并行性。我们还可以通过控制两件事来控制特定的数据布局:bucketing和partitioning(稍后讨论)。
9.8.1. 可分割的文件类型和压缩
某些文件格式基本上是“可分割的”。这可以提高速度，因为Spark可以避免读取整个文件，只访问满足查询所需的文件部分。此外，如果您正在使用Hadoop分布式文件系统(HDFS)之类的东西，如果文件跨越多个块，那么分割文件可以提供进一步的优化。与此同时，还需要管理压缩。并不是所有的压缩方案都是可分割的。如何存储数据对于使Spark作业顺利运行非常重要。我们建议gzip压缩的Parquet存储。
9.8.2. 并行读取数据
多个执行器不能同时读取同一个文件，但它们可以同时读取不同的文件。一般来说，这意味着当您从一个包含多个文件的文件夹中读取数据时，这些文件中的每一个都将成为DataFrame中的一个分区，并由可用的执行器executor并行读取(其余的文件将排在其他文件之后)
9.8.3. 并行写入数据
写入的文件或数据的数量取决于在写入数据时DataFrame拥有的分区数量。默认情况下，每个数据分区写一个文件。这意味着，尽管我们指定了一个“文件”，但它实际上是一个文件夹中的许多文件，具有指定文件的名称，每个写入的分区都有一个文件。例如，下面的代码
csvFile.repartition(5).write.format("csv").save("/tmp/multiple.csv")
将在该文件夹中生成5个文件。从列表中可以看到:
分区
分区是一种工具，允许您在编写时控制存储什么数据(以及存储在哪里)。当您将文件写入分区目录(或表)时，您基本上将列编码为文件夹。这允许您在以后读取数据时跳过大量数据，只读取与您的问题相关的数据，而不必扫描整个数据集。所有基于文件的数据源都支持这些:
写完后，你会得到一个文件夹列表在你的Parquet “文件”:
每一个都包含Parquet文件，其中包含前面谓词为true的数据:
这可能是您可以使用的最低优化，当您的表在操作之前经常被读取器过滤。例如，date对于分区特别常见，因为在下游，我们通常只想查看前一周的数据(而不是扫描整个记录列表)。这可以为读者提供大量的加速。Bucketing 分桶bucket是另一种文件组织方法，您可以使用这种方法控制专门写入每个文件的数据。这可以帮助避免以后读取数据时出现混乱，因为具有相同bucket ID的数据都将被分组到一个物理分区中。这意味着数据是根据您希望以后如何使用该数据进行预分区的，这意味着您可以在连接或聚合时避免昂贵的重新排序。与其在特定的列上进行分区(可能会写出大量目录)，还不如研究数据的分桶。这将创建一定数量的文件，并组织我们的数据到这些“桶”:
仅spark管理的表支持分桶。想了解更多关于分块和分块的信息，请观看Spark Summit 2017的这篇演讲。
https://spark-summit.org/2017/events/why-you-should-care-about-data-layout-in-the-filesystem/
9.8.4. 写复杂类型
正如我们在第6章中介绍的，Spark有各种不同的内部类型。尽管Spark可以处理所有这些类型，但并不是每种类型都能很好地处理每种数据文件格式。例如，CSV文件不支持复杂类型，而Parquet和ORC支持。
9.8.5. 管理文件的大小
管理文件大小不是写数据的重要因素，而是以后读取数据的重要因素。当您编写大量小文件时，管理所有这些文件会产生大量的元数据开销。Spark尤其不能很好地处理小文件，尽管许多文件系统(如HDFS)也不能很好地处理大量小文件。您可能会听到这被称为“小文件问题”。反过来也是正确的:您也不想要太大的文件，因为当您只需要几行数据时，读取整个数据块会变得低效。Spark 2.2引入了一种新的方法，以更自动化的方式控制文件大小。我们在前面已经看到，输出文件的数量是写时分区数量(以及我们选择的分区列)的导出数。现在，您可以利用另一个工具来限制输出文件的大小，这样您就可以获得一个最佳的文件大小。您可以使用maxRecordsPerFile选项并指定您所选择的数量。这允许您通过控制写入每个文件的记录数量来更好地控制文件大小。例如，如果为writer设置一个选项为df.write。选项(“maxRecordsPerFile”，5000)，Spark将确保文件最多包含5000条记录。
9.9. 结束语
在本章中，我们讨论了在Spark中读取和写入数据的各种选项。这几乎涵盖了您作为Spark的日常用户需要知道的所有内容。对于好奇的人来说，有几种方法可以实现您自己的数据源;但是，我们省略了如何实现这一点的说明，因为API目前正在发展为更好地支持结构化流。如果您有兴趣了解如何实现您自己的自定义数据源，那么Cassandra连接器是经过良好组织和维护的，可以为喜欢冒险的人提供参考。