数据库简介

数据库旨在将结构化数据存储为可以使用 SQL 查询读取的表。数据必须遵守严格的模式，这允许数据库管理系统对数据存储和处理进行大量协同优化。也就是说，它们将磁盘文件中的数据和索引的内部布局与其高度优化的查询处理引擎紧密结合，从而对存储的数据提供非常快速的计算，并为所有读/写操作提供强大的事务性 ACID 保证。

数据库上的 SQL 工作负载可以大致分为两类，如下所示：

在线事务处理 (OLTP) 工作负载

与银行账户事务类似，OLTP 工作负载通常是一次读取或更新几条记录的高并发、低延迟、简单的查询。

在线分析处理 (OLAP)

OLAP 工作负载，如定期报告，通常是复杂的查询（涉及聚合和连接），需要对许多记录进行高吞吐量扫描。

需要注意的是，Apache Spark 是一个主要为 OLAP 工作负载而非 OLTP 工作负载设计的查询引擎。因此，在本章的其余部分，我们将重点讨论分析工作负载的存储解决方案。接下来，让我们看看如何使用 Apache Spark 来读取和写入数据库。

使用 Apache Spark 读取和写入数据库

得益于不断增长的连接器生态系统，Apache Spark 可以连接到各种数据库以读取和写入数据。对于具有 JDBC 驱动程序的数据库（例如 PostgreSQL、MySQL），您可以使用内置的 JDBC 数据源以及适当的 JDBC 驱动程序 jar 来访问数据。对于许多其他现代数据库（例如 Azure Cosmos DB、Snowflake），您可以使用适当的格式名称调用专用连接器。第 5 章详细讨论了几个例子。这使得使用基于 Apache Spark 的工作负载和用例来扩充您的数据仓库和数据库变得非常容易。

数据库的限制

自上个世纪以来，数据库和 SQL 查询一直被认为是 BI 工作负载的出色构建解决方案。但是，在过去十年中，分析工作负载出现了两个主要的新趋势：

数据量的增长

随着大数据的出现，业界已经出现了一种全球趋势，即衡量和收集一切（页面浏览量、点击量等）以了解趋势和用户行为。结果，任何公司或组织收集的数据量都从几十年前的千兆字节增加到今天的千兆字节和千兆字节。

分析多样性的增长

随着数据收集的增加，需要更深入的洞察力。这导致了机器学习和深度学习等复杂分析的爆炸式增长。

由于以下限制，数据库已被证明在适应这些新趋势方面相当不足：

数据库的横向扩展非常昂贵

尽管数据库在单机上处理数据的效率极高，但数据量的增长速度已经远远超过单机性能能力的增长。处理引擎前进的唯一方法是横向扩展，即使用多台机器并行处理数据。但是，大多数数据库，尤其是开源数据库，并不是为横向扩展以执行分布式处理而设计的。少数能够远程满足处理要求的工业数据库解决方案往往是在专用硬件上运行的专有解决方案，因此获取和维护成本非常高。

数据库不能很好地支持基于非 SQL 的分析

数据库以复杂（通常是专有）格式存储数据，这些格式通常经过高度优化，仅供该数据库的 SQL 处理引擎读取。这意味着其他处理工具，如机器学习和深度学习系统，无法有效地访问数据（除非从数据库中低效地读取所有数据）。数据库也不能轻松扩展以执行基于非 SQL 的分析，如机器学习。

数据库的这些限制导致开发了一种完全不同的数据存储方法，称为数据湖.

数据湖简介

与数据库不同，数据湖架构将分布式存储系统与分布式计算系统解耦。这允许每个系统根据工作负载的需要进行扩展。此外，数据以开放格式保存为文件，因此任何处理引擎都可以使用标准 API 读取和写入它们。这个想法在 2000 年代后期由Apache Hadoop 项目的 Hadoop 文件系统 (HDFS) 推广，该项目本身深受 Sanjay Ghemawat、Howard Gobioff 和 Shun-Tak Leung的研究论文“The Google File System”的启发。

组织通过独立选择以下内容来构建其数据湖：

存储系统

他们选择在机器集群上运行 HDFS 或使用任何云对象存储（例如，AWS S3、Azure Data Lake Storage 或 Google Cloud Storage）。

文件格式

根据下游工作负载，数据以结构化（例如，Parquet、ORC）、半结构化（例如，JSON）或有时甚至是非结构化格式（例如，文本、图像、音频、视频）的文件形式存储。

处理引擎

同样，根据要执行的分析工作负载的种类，选择处理引擎。这可以是批处理引擎（例如，Spark、Presto、Apache Hive）、流处理引擎（例如，Spark、Apache Flink）或机器学习库（例如，Spark MLlib、scikit-learn、R）。

这种灵活性——能够选择最适合手头工作负载的存储系统、开放数据格式和处理引擎——是数据湖相对于数据库的最大优势。总体而言，对于相同的性能特征，数据湖通常提供比数据库便宜得多的解决方案。这一关键优势导致了大数据生态系统的爆炸式增长。在下一节中，我们将讨论如何使用 Apache Spark 在任何存储系统上读取和写入常见文件格式。

使用 Apache Spark 读取和写入数据湖

Apache Spark 是构建您自己的数据湖时使用的最佳处理引擎之一，因为它提供了它们所需的所有关键功能：

支持各种工作负载

Spark 提供了处理各种工作负载所需的所有工具，包括批处理、ETL 操作、使用 Spark SQL 的 SQL 工作负载、使用结构化流的流处理（在第 8 章中讨论）以及使用 MLlib 的机器学习（在第 10 章中讨论） ，等等。

支持多种文件格式

在第 4 章中，我们详细探讨了 Spark 如何内置支持非结构化、半结构化和结构化文件格式。

支持多种文件系统

Spark 支持从任何支持 Hadoop 文件系统 API 的存储系统访问数据。由于该 API 已成为大数据生态系统中的事实标准，因此大多数云和本地存储系统都为其提供了实现——这意味着 Spark 可以读取和写入大多数存储系统。

但是，对于许多文件系统（尤其是那些基于云存储的文件系统，如 AWS S3），您必须配置 Spark，使其能够以安全的方式访问文件系统。此外，云存储系统通常不具有与标准文件系统相同的文件操作语义（例如，S3 中的最终一致性），如果不相应地配置 Spark，可能会导致结果不一致。有关详细信息，请参阅有关云集成的文档。

数据湖的局限性

数据湖并非没有缺陷，其中最令人震惊的是缺乏交易保证。具体来说，数据湖无法在以下方面提供 ACID 保证：

原子性和隔离

处理引擎以分布式方式将数据写入数据湖中的多个文件。如果操作失败，则没有机制可以回滚已写入的文件，从而留下可能损坏的数据（当并发工作负载修改数据时，问题会加剧，因为如果没有更高级别的机制，很难提供跨文件的隔离） .

一致性

失败写入缺乏原子性进一步导致读者对数据的看法不一致。事实上，即使数据写入成功，也很难保证数据质量。例如，数据湖的一个非常常见的问题是意外以与现有数据不一致的格式和模式写出数据文件。

为了解决数据湖的这些限制，开发人员使用了各种技巧。这里有一些例子：

• 数据湖中的大型数据文件集合通常由基于列值的子目录“分区”（例如，按日期分区的大型 Parquet 格式的 Hive 表）。为了实现对现有数据的原子修改，通常重写整个子目录（即，写入临时目录，然后交换引用）只是为了更新或删除一些记录。

• 数据更新作业（例如，日常 ETL 作业）和数据查询作业（例如，日常报告作业）的时间表经常错开，以避免对数据的并发访问以及由此引起的任何不一致。

消除这些实际问题的尝试导致了新系统的开发，例如湖屋。

Apache Hudi

Apache Hudi最初由Uber Engineering构建，是 Hadoop Update Delete and Incremental 的首字母缩写词，是一种数据存储格式，专为增量更新插入和删除键/值样式数据而设计。数据存储为列格式（例如，Parquet 文件）和基于行的格式（例如，用于记录 Parquet 文件的增量更改的 Avro 文件）的组合。除了前面提到的常见功能外，它还支持：

• 使用快速、可插入的索引进行更新插入

• 具有回滚支持的数据的原子发布

• 读取对表的增量更改

• 数据恢复的保存点

• 使用统计的文件大小和布局管理

• 行和列数据的异步压缩

Apache Iceberg

Apache Iceberg最初由Netflix构建，是另一种用于大型数据集的开放存储格式。然而，与专注于插入键/值数据的 Hudi 不同，Iceberg 更专注于通用数据存储，可在单个表中扩展到 PB 并具有模式演变属性。具体来说，它提供了以下附加功能（除了常见的功能）：

• 通过添加、删除、更新、重命名和重新排序列、字段和/或嵌套结构来进行模式演变

• 隐藏分区，它在幕后为表中的行创建分区值

• 分区演化，它自动执行元数据操作以随着数据量或查询模式的变化更新表布局

• 时间旅行，允许您通过 ID 或时间戳查询特定的表快照

• 回滚到以前的版本以纠正错误

• 可序列化的隔离，甚至在多个并发写入者之间

Delta Lake

Delta Lake是一个由 Linux 基金会托管的开源项目，由 Apache Spark 的原始创建者构建。与其他类似，它是一种开放的数据存储格式，可提供事务保证并支持模式实施和演变。它还提供了其他一些有趣的功能，其中一些是独一无二的。Delta Lake 支持：

• 使用结构化流源和接收器对表进行流式读取和写入

• 更新、删除和合并（用于 upserts）操作，即使在 Java、Scala 和 Python API 中也是如此

• 通过显式更改表模式或通过在 DataFrame 写入期间将 DataFrame 的模式隐式合并到表的模式来进行模式演变。（事实上​​，Delta Lake 中的合并操作支持条件更新/插入/删除、一起更新所有列等高级语法，您将在本章后面看到。）

• 时间旅行，允许您通过 ID 或时间戳查询特定的表快照

• 回滚到以前的版本以纠正错误

• 执行任何 SQL、批处理或流操作的多个并发编写器之间的可序列化隔离

在本章的其余部分，我们将探讨如何使用这样的系统以及 Apache Spark 来构建提供上述属性的 Lakehouse。在这三个系统中，到目前为止，Delta Lake 与 Apache Spark 数据源（用于批处理和流式工作负载）和 SQL操作（例如，MERGE）的集成最紧密。因此，我们将使用 Delta Lake 作为进一步探索的载体。

使用 Delta Lake 配置 Apache Spark

您可以通过以下方式之一将 Apache Spark 配置为链接到 Delta Lake 库：

设置交互式外壳

如果您使用的是 Apache Spark 3.0，则可以使用以下命令行参数启动带有 Delta Lake 的 PySpark 或 Scala shell：

--packages io.delta:delta-core_2.12:0.7.0

例如：

pyspark --packages io.delta:delta-core_2.12:0.7.0

如果您运行的是 Spark 2.4，则必须使用 Delta Lake 0.6.0。

使用 Maven 坐标设置独立的 Scala/Java 项目

如果要使用 Maven 中央存储库中的 Delta Lake 二进制文件构建项目，可以将以下 Maven 坐标添加到项目依赖项中：

  <dependency>
  <groupId>io.delta</groupId>
  <artifactId>delta-core_2.12</artifactId>
  <version>0.7.0</version>
</dependency>

同样，如果您运行的是 Spark 2.4，则必须使用 Delta Lake 0.6.0。

将数据加载到 Delta Lake 表中

如果您习惯于使用 Apache Spark 和任何结构化数据格式（例如 Parquet）构建数据湖，那么迁移现有工作负载以使用 Delta Lake 格式非常容易。您所要做的就是将所有 DataFrame 读写操作更改为使用format("delta")而不是format("parquet"). 让我们用前面提到的一些贷款数据来试试这个，它可以作为 Parquet 文件使用。首先让我们读取这些数据并将其保存为 Delta Lake 表：

// In Scala
// Configure source data path
val sourcePath = "/databricks-datasets/learning-spark-v2/loans/
  loan-risks.snappy.parquet"

// Configure Delta Lake path
val deltaPath = "/tmp/loans_delta"

// Create the Delta table with the same loans data
spark
  .read
  .format("parquet")
  .load(sourcePath)
  .write
  .format("delta")
  .save(deltaPath)

// Create a view on the data called loans_delta
spark
 .read
 .format("delta")
 .load(deltaPath)
 .createOrReplaceTempView("loans_delta")

# In Python
# Configure source data path
sourcePath = "/databricks-datasets/learning-spark-v2/loans/
  loan-risks.snappy.parquet"

# Configure Delta Lake path
deltaPath = "/tmp/loans_delta"

# Create the Delta Lake table with the same loans data
(spark.read.format("parquet").load(sourcePath) 
  .write.format("delta").save(deltaPath))

# Create a view on the data called loans_delta
spark.read.format("delta").load(deltaPath).createOrReplaceTempView("loans_delta")

现在我们可以像任何其他表格一样轻松地阅读和探索数据：

// In Scala/Python

// Loans row count
spark.sql("SELECT count(*) FROM loans_delta").show()

+--------+
|count(1)|
+--------+
|   14705|
+--------+

// First 5 rows of loans table
spark.sql("SELECT * FROM loans_delta LIMIT 5").show()

+-------+-----------+---------+----------+
|loan_id|funded_amnt|paid_amnt|addr_state|
+-------+-----------+---------+----------+
|      0|       1000|   182.22|        CA|
|      1|       1000|   361.19|        WA|
|      2|       1000|   176.26|        TX|
|      3|       1000|   1000.0|        OK|
|      4|       1000|   249.98|        PA|
+-------+-----------+---------+----------+

将数据流加载到 Delta Lake 表中

与静态 DataFrame 一样，您可以通过将格式设置为"delta". 假设您有一个名为 DataFrame 的新贷款数据流newLoanStreamDF，它与表具有相同的架构。您可以按如下方式附加到表中：

// In Scala
import org.apache.spark.sql.streaming._

val newLoanStreamDF = ...   // Streaming DataFrame with new loans data
val checkpointDir = ...     // Directory for streaming checkpoints
val streamingQuery = newLoanStreamDF.writeStream
  .format("delta")
  .option("checkpointLocation", checkpointDir)
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .start(deltaPath)

# In Python
newLoanStreamDF = ...   # Streaming DataFrame with new loans data
checkpointDir = ...     # Directory for streaming checkpoints
streamingQuery = (newLoanStreamDF.writeStream 
    .format("delta") 
    .option("checkpointLocation", checkpointDir) 
    .trigger(processingTime = "10 seconds") 
    .start(deltaPath))

使用这种格式，就像任何其他格式一样，结构化流提供端到端的精确一次保证。但是，与 JSON、Parquet 或 ORC 等传统格式相比，Delta Lake 具有一些额外的优势：

它允许从批处理和流式作业写入同一个表

对于其他格式，从结构化流作业写入表的数据将覆盖表中的任何现有数据。这是因为在表中维护的元数据以确保流式写入的精确一次保证不考虑其他非流式写入。Delta Lake 的高级元数据管理允许写入批处理和流式数据。

它允许多个流作业将数据附加到同一个表

元数据与其他格式的相同限制也阻止了多个结构化流查询附加到同一个表。Delta Lake 的元数据维护每个流式查询的事务信息，从而使任意数量的流式查询能够同时写入一个表，并保证一次性。

即使在并发写入的情况下，它也提供 ACID 保证

与内置格式不同，Delta Lake 允许并发批处理和流式操作来写入具有 ACID保证的数据。

在写入时强制执行架构以防止数据损坏

使用 JSON、Parquet 和 ORC 等常见格式使用 Spark 管理数据的一个常见问题是由于写入格式不正确的数据而导致的意外数据损坏。由于这些格式定义了单个文件而不是整个表的数据布局，因此没有机制可以阻止任何 Spark 作业将具有不同模式的文件写入现有表。这意味着无法保证许多 Parquet 文件的整个表的一致性。

Delta Lake 格式将架构记录为表级元数据。因此，对 Delta Lake 表的所有写入都可以验证正在写入的数据是否具有与表兼容的模式。如果不兼容，Spark 将在任何数据写入并提交到表之前抛出错误，从而防止这种意外的数据损坏。让我们通过尝试写入一些带有附加列的数据来测试这一点closed， 表示贷款是否已终止。请注意，该列在表中不存在：

// In Scala
val loanUpdates = Seq(
    (1111111L, 1000, 1000.0, "TX", false), 
    (2222222L, 2000, 0.0, "CA", true))
  .toDF("loan_id", "funded_amnt", "paid_amnt", "addr_state", "closed")
  
loanUpdates.write.format("delta").mode("append").save(deltaPath)

# In Python
from pyspark.sql.functions import *

cols = ['loan_id', 'funded_amnt', 'paid_amnt', 'addr_state', 'closed']
items = [
(1111111, 1000, 1000.0, 'TX', True), 
(2222222, 2000, 0.0, 'CA', False)
]

loanUpdates = (spark.createDataFrame(items, cols)
  .withColumn("funded_amnt", col("funded_amnt").cast("int")))
loanUpdates.write.format("delta").mode("append").save(deltaPath)

此写入将失败并显示以下错误消息：

org.apache.spark.sql.AnalysisException: A schema mismatch detected when writing 
  to the Delta table (Table ID: 48bfa949-5a09-49ce-96cb-34090ab7d695).
To enable schema migration, please set:
'.option("mergeSchema", "true")'.

Table schema:
root
-- loan_id: long (nullable = true)
-- funded_amnt: integer (nullable = true)
-- paid_amnt: double (nullable = true)
-- addr_state: string (nullable = true)


Data schema:
root
-- loan_id: long (nullable = true)
-- funded_amnt: integer (nullable = true)
-- paid_amnt: double (nullable = true)
-- addr_state: string (nullable = true)
-- closed: boolean (nullable = true)

这说明了 Delta Lake 如何阻止与表架构不匹配的写入。但是，它也提示了如何使用 option 实际演变表的模式mergeSchema，如下所述.

不断发展的模式以适应不断变化的数据

在我们这个不断变化的数据世界中，我们可能希望将这个新列添加到表中。这个新列可以通过将选项设置为显式"mergeSchema"添加"true"：

// In Scala
loanUpdates.write.format("delta").mode("append")
  .option("mergeSchema", "true")
  .save(deltaPath)

# In Python
(loanUpdates.write.format("delta").mode("append")
  .option("mergeSchema", "true")
  .save(deltaPath))

这样，该列closed将被添加到表模式中，并且将附加新数据。读取现有行时，新列的值被视为NULL。在 Spark 3.0 中，您还可以使用 SQL DDL 命令ALTER TABLE添加和修改列。

转换现有数据

Delta Lake 支持 DML 命令UPDATE、DELETE和MERGE，它们允许您构建复杂的数据管道。可以使用 Java、Scala、Python 和 SQL 调用这些命令，从而使用户可以灵活地将命令与他们熟悉的任何 API 一起使用，使用 DataFrames 或表。此外，这些数据修改操作中的每一个都确保了 ACID 保证。

让我们通过一些实际用例的例子来探索这一点。

// In Scala
import io.delta.tables.DeltaTable
import org.apache.spark.sql.functions._

val deltaTable = DeltaTable.forPath(spark, deltaPath)
deltaTable.update(
  col("addr_state") === "OR",
  Map("addr_state" -> lit("WA")))

# In Python
from delta.tables import *

deltaTable = DeltaTable.forPath(spark, deltaPath)
deltaTable.update("addr_state = 'OR'",  {"addr_state": "'WA'"})

// In Scala
val deltaTable = DeltaTable.forPath(spark, deltaPath)
deltaTable.delete("funded_amnt >= paid_amnt")

# In Python
deltaTable = DeltaTable.forPath(spark, deltaPath)
deltaTable.delete("funded_amnt >= paid_amnt")

// In Scala
deltaTable
  .alias("t")
  .merge(loanUpdates.alias("s"), "t.loan_id = s.loan_id")
  .whenMatched.updateAll()
  .whenNotMatched.insertAll()
  .execute()

# In Python
(deltaTable
  .alias("t")
  .merge(loanUpdates.alias("s"), "t.loan_id = s.loan_id") 
  .whenMatchedUpdateAll() 
  .whenNotMatchedInsertAll() 
  .execute())

// In Scala
deltaTable
  .alias("t")
  .merge(historicalUpdates.alias("s"), "t.loan_id = s.loan_id")
  .whenNotMatched.insertAll()
  .execute()

# In Python
(deltaTable
  .alias("t")
  .merge(historicalUpdates.alias("s"), "t.loan_id = s.loan_id") 
  .whenNotMatchedInsertAll() 
  .execute())

使用操作历史审计数据更改

对 Delta Lake 表的所有更改都将作为提交记录在表的事务日志中。当您写入 Delta Lake 表或目录时，每个操作都会自动进行版本控制。您可以查询表的操作历史记录，如以下代码段所述：

// In Scala/PythondeltaTable.history().show()

默认情况下，这将显示一个包含许多版本和许多列的巨大表。让我们打印最后三个操作的一些关键列：

// In Scala
deltaTable
  .history(3)
  .select("version", "timestamp", "operation", "operationParameters")
  .show(false)

# In Python
(deltaTable
  .history(3)
  .select("version", "timestamp", "operation", "operationParameters")
  .show(truncate=False))

这将生成以下输出：

+-------+-----------+---------+-------------------------------------------+
|version|timestamp  |operation|operationParameters                        |
+-------+-----------+---------+-------------------------------------------+
|5      |2020-04-07 |MERGE    |[predicate -> (t.`loan_id` = s.`loan_id`)] |
|4      |2020-04-07 |MERGE    |[predicate -> (t.`loan_id` = s.`loan_id`)] |
|3      |2020-04-07 |DELETE   |[predicate -> ["(CAST(`funded_amnt` ...    |
+-------+-----------+---------+-------------------------------------------+

请注意operationandoperationParameters对审核更改很有用。

使用时间旅行查询表的先前快照

您可以使用DataFrameReader选项"versionAsOf"和查询表的先前版本快照"timestampAsOf"。这里有一些例子：

// In Scala
spark.read
  .format("delta")
  .option("timestampAsOf", "2020-01-01")  // timestamp after table creation
  .load(deltaPath)

spark.read.format("delta")
  .option("versionAsOf", "4")
  .load(deltaPath)

# In Python
(spark.read
  .format("delta")
  .option("timestampAsOf", "2020-01-01")  # timestamp after table creation
  .load(deltaPath))

(spark.read.format("delta")
  .option("versionAsOf", "4")
  .load(deltaPath))

这在各种情况下都很有用，例如:

• 通过在特定表版本上重新运行作业来重现机器学习实验和报告

• 比较不同版本之间的数据更改以进行审计

• 通过读取以前的快照作为 DataFrame 并用它覆盖表来回滚不正确的更改