【Apache Spark 】第 9 章使用 Apache Spark构建可靠的数据湖

🔎大家好，我是Sonhhxg_柒，希望你看完之后，能对你有所帮助，不足请指正！共同学习交流🔎

📝个人主页－Sonhhxg_柒的博客_CSDN博客 📃

🎁欢迎各位→点赞👍 + 收藏⭐️ + 留言📝​

📣系列专栏 - 机器学习【ML】 自然语言处理【NLP】 深度学习【DL】

​

🖍foreword

✔说明⇢本人讲解主要包括Python、机器学习（ML）、深度学习（DL）、自然语言处理（NLP）等内容。

如果你对这个系列感兴趣的话，可以关注订阅哟👋

在前面的章节中，您学习了如何轻松有效地使用 Apache Spark 构建可扩展的高性能数据处理管道。然而，在实践中，表达处理逻辑只解决了构建管道的端到端问题的一半。对于数据工程师、数据科学家或数据分析师来说，构建管道的最终目标是查询处理过的数据并从中获得洞察力。存储解决方案的选择决定了数据管道的端到端（即从原始数据到洞察）的稳健性和性能。

在本章中，我们将首先讨论您需要注意的存储解决方案的关键特性。然后我们将讨论两大类存储解决方案、数据库和数据湖，以及如何将 Apache Spark 与它们一起使用。最后，我们将介绍下一波存储解决方案，称为 Lakehouses，并探索该领域的一些新的开源处理引擎。

最佳存储解决方案的重要性

以下是存储解决方案中所需的一些属性：

可扩展性和性能

存储解决方案应该能够根据数据量进行扩展，并提供工作负载所需的读/写吞吐量和延迟。

交易支持

复杂的工作负载通常是同时读写数据，因此对ACID 事务的支持对于确保最终结果的质量至关重要。

支持多种数据格式

存储解决方案应该能够存储非结构化数据（例如，原始日志等文本文件）、半结构化数据（例如，JSON 数据）和结构化数据（例如，表格数据）。

支持各种工作负载

存储解决方案应该能够支持各种业务工作负载，包括：

• SQL 工作负载，如传统 BI 分析
• 批处理工作负载，例如处理原始非结构化数据的传统 ETL 作业
• 流式传输工作负载，例如实时监控和警报
• ML 和 AI 工作负载，例如推荐和流失预测

开放性

支持广泛的工作负载通常需要以开放数据格式存储数据。标准 API 允许从各种工具和引擎访问数据。这允许企业针对每种类型的工作负载使用最优化的工具，并做出最佳的业务决策。

随着时间的推移，已经提出了不同种类的存储解决方案，每种存储解决方案在这些属性方面都有其独特的优点和缺点。在本章中，我们将探讨可用的存储解决方案如何从数据库演变为数据湖，以及如何将 Apache Spark 与它们一起使用。然后，我们将把注意力转向下一代存储解决方案，通常称为数据湖库，它可以提供两全其美的优势：数据湖的可扩展性和灵活性以及数据库的事务保证。

数据库

几十年来，数据库一直是构建数据仓库以存储关键业务数据的最可靠解决方案。在本节中，我们将探讨数据库的架构及其工作负载，以及如何使用 Apache Spark 处理数据库上的分析工作负载。我们将在本节结束时讨论数据库在支持现代非 SQL 工作负载方面的局限性。

数据库简介

数据库旨在将结构化数据存储为可以使用 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 的分析，如机器学习。

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

数据湖

与大多数数据库相比，数据湖是一种分布式存储解决方案，可在商用硬件上运行并轻松横向扩展。在本节中，我们将首先讨论数据湖如何满足现代工作负载的要求，然后了解 Apache Spark 如何与数据湖集成以使工作负载扩展到任何规模的数据。最后，我们将探讨数据湖为实现可扩展性而做出的架构牺牲的影响。

数据湖简介

与数据库不同，数据湖架构将分布式存储系统与分布式计算系统解耦。这允许每个系统根据工作负载的需要进行扩展。此外，数据以开放格式保存为文件，因此任何处理引擎都可以使用标准 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 作业）和数据查询作业（例如，日常报告作业）的时间表经常错开，以避免对数据的并发访问以及由此引起的任何不一致。

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

Lakehouses：存储解决方案演进的下一步

Lakehouse是一种新的范例，它结合了数据湖和数据仓库的最佳元素，用于 OLAP 工作负载。Lakehouses 是通过一种新的系统设计实现的，该设计直接在用于数据湖的低成本、可扩展存储上提供类似于数据库的数据管理功能。更具体地说，它们提供以下功能：

交易支持

与数据库类似，Lakehouse 在存在并发工作负载的情况下提供 ACID 保证。

模式执行和治理

Lakehouses 可防止将具有不正确模式的数据插入到表中，并且在需要时，可以显式地改进表模式以适应不断变化的数据。系统应该能够推理数据完整性，并且应该具有强大的治理和审计机制。

支持开放格式的多种数据类型

与数据库不同，但类似于数据湖，Lakehouse 可以存储、优化、分析和访问许多新数据应用程序所需的所有类型的数据，无论是结构化、半结构化还是非结构化数据。为了使各种工具能够直接有效地访问它，数据必须以开放格式存储，并使用标准化的 API 来读取和写入它们。

支持各种工作负载

在使用开放 API 读取数据的各种工具的支持下，Lakehouses 使不同的工作负载能够对单个存储库中的数据进行操作。打破孤立的数据孤岛（即，不同类别数据的多个存储库）使开发人员能够更轻松地构建多样化和复杂的数据解决方案，从传统的 SQL 和流式分析到机器学习。

支持 upserts 和删除

更改数据捕获 (CDC)和渐变维度 (SCD)操作等复杂用例需要不断更新表中的数据。Lakehouses 允许通过事务保证同时删除和更新数据。

数据治理

Lakehouses 提供了一些工具，您可以使用这些工具来推断数据完整性并审核所有数据更改以确保合规性。

目前，有一些开源系统，例如 Apache Hudi、Apache Iceberg 和 Delta Lake，可用于构建具有这些属性的湖屋。在非常高的层次上，这三个项目都具有相似的架构，其灵感来自于众所周知的数据库原则。它们都是执行以下操作的开放数据存储格式：

• 在可扩展的文件系统上以结构化文件格式存储大量数据。
• 维护事务日志以记录数据原子更改的时间线（很像数据库）。
• 使用日志来定义表数据的版本，并在读写器之间提供快照隔离保证。
• 支持使用 Apache Spark 读写表。

在这些广泛的范围内，每个项目在 API、性能以及与 Apache Spark 的数据源 API 的集成级别方面都具有独特的特征。接下来我们将探索它们。请注意，所有这些项目都在快速发展，因此某些描述在您阅读它们时可能已经过时。有关最新信息，请参阅每个项目的在线文档。

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，因为它类似于流媒体。溪流流入大海形成三角洲——这是所有沉积物堆积的地方，也是有价值的农作物生长的地方。Jules S. Damji（我们的合著者之一）想出了这个！

使用 Apache Spark 和 Delta Lake 构建 Lakehouses

在本节中，我们将快速了解如何使用 Delta Lake 和 Apache Spark 来构建 Lakehouse。具体来说，我们将探讨以下内容：

• 使用 Apache Spark 读取和写入 Delta Lake 表
• Delta Lake 如何通过 ACID保证允许并发批处理和流式写入
• Delta Lake 如何通过对所有写入强制执行模式来确保更好的数据质量，同时允许显式模式演变
• 使用更新、删除和合并操作构建复杂的数据管道，所有这些都确保了 ACID 保证
• 审核修改 Delta Lake 表的操作的历史记录，并通过查询表的早期版本及时返回

我们将在本节中使用的数据是公共Lending Club Loan Data的修改版本（Parquet 格式的列的子集） 。1包括 2012 年至 2017 年所有资助贷款。每笔贷款记录包括申请人提供的申请人信息以及当前的贷款状态（当前、逾期、全额支付等）和最近的付款信息。

使用 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 文档。

将数据加载到 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 保证。

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

更新数据以修复错误

管理数据时的一个常见用例是修复数据中的错误。假设，在查看数据后，我们意识到分配给的所有贷款都

addr_state = 'OR'

应该分配给

addr_state = 'WA'

。如果贷款表是 Parquet 表，那么要进行这样的更新，我们需要：

1. 将所有不受影响的行复制到新表中。
2. 将所有受影响的行复制到 DataFrame 中，然后执行数据修改。
3. 将前面提到的 DataFrame 的行插入到新表中。
4. 删除旧表并将新表重命名为旧表名。

UPDATE

在 Spark 3.0 中，增加了对 DML SQL 操作（如、

DELETE

和）的直接支持

MERGE

，您无需手动执行所有这些步骤，只需运行 SQL

UPDATE

命令即可。但是，对于 Delta Lake 表，用户也可以通过使用 Delta Lake 的编程 API 来运行此操作，如下所示：

// 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'"})

删除用户相关数据

随着欧盟通用数据保护条例 (GDPR)等数据保护政策的生效，现在比以往任何时候都更重要的是能够从所有表中删除用户数据。假设您必须删除所有已全额还清的贷款的数据。使用 Delta Lake，您可以执行以下操作：

// 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")

与更新类似，使用 Delta Lake 和 Apache Spark 3.0，您可以直接

DELETE

在表上运行 SQL 命令。

使用 merge() 将更改数据更新到表中

一个常见的用例是更改数据捕获，您必须将 OLTP 表中的行更改复制到另一个表以用于 OLAP 工作负载。继续我们的贷款数据示例，假设我们有另一个新贷款信息表，其中一些是新贷款，另一些是对现有贷款的更新。此外，假设该表与该

changes

表具有相同的架构

loan_delta

。

DeltaTable.merge()

您可以使用基于

MERGE

SQL 命令的操作将这些更改插入到表中：

// 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())

MERGE

提醒一下，您可以从 Spark 3.0 开始将其作为 SQL 命令运行。此外，如果您有此类捕获的更改流，则可以使用结构化流查询持续应用这些更改。该查询可以从任何流式源中读取微批次的变化（参见第 8 章

foreachBatch()

），并用于将每个微批次中的变化应用到 Delta Lake 表中。

使用仅插入合并插入时删除重复数据

Delta Lake 中的合并操作支持超出 ANSI 标准指定的扩展语法，包括以下高级功能：

删除操作

例如，

MERGE ... WHEN MATCHED THEN DELETE

。

条款条件

例如，。

MERGE ... WHEN MATCHED AND *<condition>* THEN ...

可选操作

所有

MATCHED

and

NOT MATCHED

子句都是可选的。

星形语法

例如，

UPDATE *

使用

INSERT *

源数据集中的匹配列更新/插入目标表中的所有列。等效的 Delta Lake API 是

updateAll()

和

insertAll()

，我们在上一节中看到。

这使您可以用很少的代码表达许多更复杂的用例。例如，假设您想

loan_delta

用过去贷款的历史数据回填表格。但是某些历史数据可能已经插入到表中，您不想更新这些记录，因为它们可能包含更多最新信息。您可以通过

loan_id

仅使用操作运行以下合并操作来在插入时进行重复数据删除

INSERT

（因为该

UPDATE

操作是可选的）：

// 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())

还有更复杂的用例，例如带有删除和 SCD 表的 CDC，它们通过扩展的合并语法变得简单。有关更多详细信息和示例，请参阅文档.

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

对 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` ...    |
+-------+-----------+---------+-------------------------------------------+

请注意

operation

and

operationParameters

对审核更改很有用。

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

您可以使用

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 并用它覆盖表来回滚不正确的更改

概括

本章探讨了使用 Apache Spark 构建可靠数据湖的可能性。回顾一下，数据库已经解决了很长时间的数据问题，但它们无法满足现代用例和工作负载的多样化需求。构建数据湖是为了缓解数据库的一些限制，而 Apache Spark 是构建它们的最佳工具之一。然而，数据湖仍然缺乏数据库提供的一些关键特性（例如，ACID 保证）。Lakehouses 是下一代数据解决方案，旨在提供数据库和数据湖的最佳功能，并满足各种用例和工作负载的所有要求。

我们简要探讨了一些可用于构建 Lakehouse 的开源系统（Apache Hudi 和 Apache Iceberg），然后仔细研究了 Delta Lake，这是一种基于文件的开源存储格式，与 Apache Spark 一起，是湖屋的伟大基石。如您所见，它提供以下内容：

• 事务保证和模式管理，如数据库
• 可扩展性和开放性，如数据湖
• 支持具有 ACID 保证的并发批处理和流式工作负载
• 支持使用确保 ACID 保证的更新、删除和合并操作转换现有数据
• 支持版本控制、操作历史审计、历史版本查询