Delta Lake 时间旅行
这篇博文展示了如何使用 Delta Lake 在 Delta 表的不同版本之间进行时间旅行。你可以按表版本或按时间戳进行时间旅行。你将了解时间旅行的好处，以及为什么它是生产数据工作负载必不可少的功能。
你还将了解 vacuum 命令如何限制你的时间旅行能力。你必须制定一个符合你的时间旅行需求的最佳 vacuum 策略。
让我们深入到一个例子中。如果你想跟着做，所有代码片段都在此笔记本中。
Delta Lake 按版本进行时间旅行的 Python 示例
让我们创建一个包含以下三个版本的 Delta 表
我们将通过创建 Delta 表、追加一些数据，然后执行覆盖操作来构建包含这三个版本的 Delta 表。
让我们首先创建 Delta 表以创建版本 0
df = spark.range(0, 3)
df.repartition(1).write.format("delta").save("tmp/some_nums")注意：我们使用 repartition(1) 输出单个文件以简化此演示。通常不应输出单个文件。
现在将一些数据追加到 Delta 表中，这将创建版本 1
df = spark.range(8, 11)
df.repartition(1).write.mode("append").format("delta").save("tmp/some_nums")最后，覆盖 Delta 表，这将创建 Delta 表的版本 2
df = spark.createDataFrame([(55,), (66,), (77,)]).toDF("id")
df.repartition(1).write.mode("overwrite").format("delta").save("tmp/some_nums")让我们读取 Delta 表的最新版本以确认它只包含版本 2 数据
spark.read.format("delta").load("tmp/some_nums").show()
+---+
| id|
+---+
| 55|
| 66|
| 77|
+---+让我们看一些时间旅行到不同版本数据的例子。以下是如何时间旅行回到版本 0 并读取 Delta 表的早期版本
spark.read.format("delta").option("versionAsOf", "0").load("tmp/some_nums").show()
+---+
| id|
+---+
| 0|
| 1|
| 2|
+---+现在读取 Delta 表的版本 1
spark.read.format("delta").option("versionAsOf", "1").load("tmp/some_nums").show()
+---+
| id|
+---+
| 8|
| 9|
| 10|
| 0|
| 1|
| 2|
+---+我们已经看到，当未显式设置 versionAsOf 时，Delta Lake 默认会读取表的最新版本。你也可以显式读取 Delta 表的最新版本（在这种情况下是版本 2）
spark.read.format("delta").option("versionAsOf", "2").load("tmp/some_nums").show()
+---+
| id|
+---+
| 55|
| 66|
| 77|
+---+Delta Lake 使得时间旅行和读取 Delta 表的不同版本变得容易。你可以使用 history 命令来显示 Delta 表的所有版本以及关联的时间戳。
from delta.tables import DeltaTable
delta_table = DeltaTable.forPath(spark, "tmp/some_nums")
delta_table.history().select("version", "timestamp", "operation").show(truncate=False)这是显示的表格
+-------+-----------------------+---------+
|version|timestamp |operation|
+-------+-----------------------+---------+
|2 |2023-01-31 06:27:08.508|WRITE |
|1 |2023-01-31 06:26:56.895|WRITE |
|0 |2023-01-31 06:26:35.825|WRITE |
+-------+-----------------------+---------+让我们进一步了解 Delta Lake 是如何构建以实现此功能的。
Delta Lake 时间旅行直觉
Delta Lake 将数据存储在 Parquet 文件中，并将事务信息存储在 _delta_log 元数据文件夹中。
_delta_log 元数据文件夹跟踪每个事务从 Delta 表中添加和删除的 Parquet 数据文件。
下图显示了我们示例中每个事务添加和删除的文件。
让我们看看 Delta Lake 如何检查事务日志并确定每个版本应该读取哪些文件
- 对于版本 0，Delta Lake 只需读取文件 A
- Delta Lake 将看到版本 1 应读取文件 A 和文件 B
- 对于版本 2，Delta Lake 将看到文件 A、文件 B 和文件 C 已添加，但文件 A 和文件 B 已删除，因此只应读取文件 C。在读取版本 2 时，Delta Lake 将只读取文件 C 并跳过其他文件。
Delta Lake 通过查阅事务日志智能地读取文件。查询引擎在读取存储在数据湖中的文件时需要执行昂贵的文件列表操作。智能地查询事务日志以获取给定版本的相关文件效率更高。
Delta Lake 按时间戳进行时间旅行
Delta Lake 还允许你根据时间戳进行时间旅行。
spark.read.format("delta").option("timestampAsOf", "2019-01-01").load("tmp/some_nums")按时间戳进行时间旅行是访问早期数据状态而无需找出确切版本的好方法。
如果你根据时间戳进行时间旅行并将 Delta 表复制到新位置，则需要了解一个重要的细节。Delta Lake 基于时间的旅行依赖于文件时间戳。当你将 Delta Lake 表复制到另一个位置时，文件时间戳可能会更改，这将更改你的基于时间的时间旅行代码的行为。因此，在将 Delta 表复制到另一个位置时，保留现有文件时间戳至关重要。
当然，如果你按版本号进行时间旅行，复制文件时更改时间戳不是你需要考虑的因素。
Delta Lake 时间旅行 SQL 示例
这是时间旅行到特定版本的 SQL 语法
SELECT count(*) FROM my_table VERSION AS OF 5238这是时间旅行到特定时间戳的 SQL 语法
SELECT count(*) FROM my_table TIMESTAMP AS OF "2019-01-01"无论你使用 SQL 还是 Python API，时间旅行的工作方式都相同。唯一的区别是语法。
vacuum 后的 Delta Lake 时间旅行
Delta Lake 支持 VACUUM 命令，该命令从存储中删除早于保留期且在事务日志中标记为删除的文件。
如果 VACUUM 命令从存储中删除了给定 Delta 表版本依赖的文件，你将无法再时间旅行到该 Delta 表版本。
你需要根据给定表的时间旅行需求设置 Delta 表的保留期。某些表永远不应进行 vacuum，因为你希望保留对每个版本的访问。其他表应经常进行 vacuum 以节省存储成本。有关更多详细信息，请参阅这篇关于 vacuum 命令的博文。
你应该确保为你的表设置一个针对你的时间旅行需求进行优化的 vacuum 策略。
Delta Lake 恢复与时间旅行
Delta Lake 支持 restore 命令，该命令可以轻松地将 Delta 表的先前版本恢复为“当前版本”。有关更多信息，请参阅这篇关于 RESTORE 命令的博文。
每次时间旅行时都必须指定版本，因为它不像 RESTORE 那样更改 Delta 表的当前版本。假设一个 Delta 表有三个版本，你时间旅行回版本 0。当你再次读取 Delta 表的最新版本时，即使你已经时间旅行了，它仍然会默认返回到当前版本。
当你想要重置表的当前版本时，RESTORE 更好。例如，当你摄入错误数据并希望撤消其他读取器的追加时，restore 非常有用。
Delta Lake 时间旅行与数据湖支持
Delta Lake 使得在 Delta 表的不同版本之间进行时间旅行变得容易。这是一个直接的操作，是 Delta Lake 事务日志的自然扩展。
数据湖不支持时间旅行。当你读取数据湖时，你总是必须读取最新版本。内置的时间旅行是 Delta Lake 与数据湖相比提供的重要功能。
结论
这篇博文教你如何按版本号和时间戳在 Delta 表的不同版本之间进行时间旅行。
你了解了 Delta 事务日志如何轻松实现时间旅行，以及删除 Delta 表中的历史文件如何限制你时间旅行到特定版本的能力。
你需要仔细设置 Delta 表的保留期，以便你可以进行时间旅行，同时具有足够的灵活性来 vacuum 文件并节省存储成本（如果这很重要）。有关更多详细信息，请参阅vacuum 博文。
请参阅此处关于时间旅行的另一篇博文，其中描述了此功能的实际用例。
上关注我们的作者