使用表历史记录

修改表的每个操作都会创建一个新的表版本。 使用历史记录信息审核操作、回滚表或使用时间旅行在特定时间点查询表。

检索表历史记录

通过运行 history 命令，检索有关每次表写入的操作、用户和时间戳等信息。 操作按时间倒序返回。

表历史记录保留期取决于表设置 logRetentionDuration，后者默认为 30 天。

DESCRIBE HISTORY table_name       -- get the full history of the table

DESCRIBE HISTORY table_name LIMIT 1  -- get the last operation only

有关 Spark SQL 语法的详细信息，请参阅 DESCRIBE HISTORY。

请参阅 Delta Lake API 文档，了解 Scala/Java/Python 语法详细信息。

目录资源管理器 提供此详细表信息和历史记录的可视视图。 除了表架构和示例数据之外，你还可以单击“历史记录”选项卡以查看随 一起显示的表历史记录。

历史模式

history 操作的输出包含下列列。

+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|version|          timestamp|userId|userName|operation| operationParameters| job|notebook|clusterId|readVersion|   isolationLevel|isBlindAppend|    operationMetrics|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|      5|2019-07-29 14:07:47|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          4|WriteSerializable|        false|[numTotalRows -> ...|
|      4|2019-07-29 14:07:41|   ###|     ###|   UPDATE|[predicate -> (id...|null|     ###|      ###|          3|WriteSerializable|        false|[numTotalRows -> ...|
|      3|2019-07-29 14:07:29|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          2|WriteSerializable|        false|[numTotalRows -> ...|
|      2|2019-07-29 14:06:56|   ###|     ###|   UPDATE|[predicate -> (id...|null|     ###|      ###|          1|WriteSerializable|        false|[numTotalRows -> ...|
|      1|2019-07-29 14:04:31|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          0|WriteSerializable|        false|[numTotalRows -> ...|
|      0|2019-07-29 14:01:40|   ###|     ###|    WRITE|[mode -> ErrorIfE...|null|     ###|      ###|       null|WriteSerializable|         true|[numFiles -> 2, n...|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+

了解 partitionBy 在操作参数中的应用

该 partitionBy 字段仅在定义或更改表分区架构的 CREATE 和 OVERWRITE 操作中才有意义。

对于现有表（APPEND、INSERT、UPDATE、DELETE、MERGE）的追加作，此字段可能会显示空数组 [] 或分区列，具体取决于所使用的写入方法（.save() vs .saveAsTable()）。 此不一致是预期行为，不应用于验证写入。

例子

请考虑按 date 列分区的表：

# Initial table creation - partitionBy is populated
df.write.format("delta") \
  .partitionBy("date") \
  .saveAsTable("sales_data")

历史记录中显示的 CREATE操作为：

operationParameters: {
  "mode": "ErrorIfExists",
  "partitionBy": "[\"date\"]"
}

将数据追加到此表时：

# Subsequent append - partitionBy shows empty
new_df.write.format("delta") \
  .mode("append") \
  .saveAsTable("sales_data")

APPEND操作显示：

operationParameters: {
  "mode": "Append",
  "partitionBy": "[]"
}

partitionBy 的空值是预期的。 数据仍会根据表的现有分区架构写入正确的分区。 请注意，.save() 路径可能在此字段中显示分区列，但此差异属于实现细节，不会影响写入行为。

操作指标

history 操作返回 operationMetrics 列映射中操作指标的集合。

下表按操作将映射键的定义列出。

WRITE, CREATE TABLE AS SELECT, REPLACE TABLE AS SELECT, COPY INTO

以下指标可用于这些操作：

STREAMING UPDATE

以下指标可用于此操作：

DELETE

以下指标可用于此操作：

TRUNCATE

以下指标可用于此操作：

MERGE

以下指标可用于此操作：

UPDATE

以下指标可用于此操作：

FSCK

以下指标可用于此操作：

CONVERT

以下指标可用于此操作：

OPTIMIZE

以下指标可用于此操作：

CLONE

以下指标可用于此操作：

RESTORE

以下指标可用于此操作：

VACUUM

以下指标可用于此操作：

什么是时间旅行？

时间旅行支持根据时间戳或表版本（如事务日志中记录）查询以前的表版本。 可以使用时间旅行应用于以下情况：

• 重新创建分析、报表或输出（例如，机器学习模型的输出）。 这对于调试或审核非常有用，尤其是在管控行业中。
• 编写复杂的时态查询。
• 修复数据中的错误。
• 为针对快速变化表的一组查询提供快照隔离。

时间旅行语法

通过在表名定义之后添加子句来查询支持时间旅行功能的表。

• timestamp_expression 可以是下列项中的任意一项：
  • '2018-10-18T22:15:12.013Z'，即可以转换为时间戳的字符串
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18'，即日期字符串
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • 能够转换为时间戳或本身就是时间戳的任何其他表达式
• version 是可以从 DESCRIBE HISTORY table_spec 的输出中获取的 long 值。

timestamp_expression 和 version 都不能是子查询。

只接受日期或时间戳字符串。 例如，"2019-01-01" 和 "2019-01-01T00:00:00.000Z"。 有关示例语法，请参阅以下代码：

SQL

SELECT * FROM people10m TIMESTAMP AS OF '2018-10-18T22:15:12.013Z';
SELECT * FROM people10m VERSION AS OF 123;

Python

df1 = spark.read.option("timestampAsOf", "2019-01-01").table("people10m")
df2 = spark.read.option("versionAsOf", 123).table("people10m")

还可以使用 @ 语法将时间戳或版本指定为表名称的一部分。 时间戳必须采用 yyyyMMddHHmmssSSS 格式。 你可以通过在 @ 后附加一个 v 来指定版本。 有关示例语法，请参阅以下代码：

SQL

SELECT * FROM people10m@20190101000000000
SELECT * FROM people10m@v123

Python

spark.read.table("people10m@20190101000000000")
spark.read.table("people10m@v123")

什么是事务日志检查点？

表版本记录为事务日志目录中的 JSON 文件，该目录与表数据一起存储。 为了优化检查点查询，表版本会聚合到 Parquet 检查点文件，从而防止需要读取表历史记录的所有 JSON 版本。 Azure Databricks 针对数据大小和工作负荷优化检查点操作频率。 用户应该不需要直接与检查点交互。 检查点频率可能会更改，恕不另行通知。

为“时光旅行查询”配置数据保留

若要查询以前的表版本，必须同时保留该版本的日志文件和数据文件。

当 VACUUM 针对某个表运行时，会删除数据文件。 日志文件删除在检查点表版本后自动进行管理。

由于大多数表 VACUUM 都定期针对它们运行，因此时间点查询应遵循保留阈值 VACUUM，默认情况下为 7 天。

若要增加表的数据保留阈值，必须配置下表属性：

• delta.logRetentionDuration = "interval <interval>"：控制表的历史记录的保留时间长度。 默认值为 interval 30 days。 对于 Apache Iceberg 表，请改用 iceberg.logRetentionDuration 。
• delta.deletedFileRetentionDuration = "interval <interval>"：确定由 VACUUM 用来删除当前表版本中不再引用的数据文件的阈值。 默认值为 interval 7 days。 对于 Apache Iceberg 表，请改用 iceberg.deletedFileRetentionDuration 。

可以在创建表期间指定表属性，也可以使用语句设置它们 ALTER TABLE 。 请参阅 表属性参考。

将表还原到早期状态

可以使用命令将表还原到其早期状态 RESTORE 。 表内部维护历史版本，使它们能够还原到先前状态。 RESTORE 命令支持使用一个与早期状态相对应的版本作为选项，或支持使用一个表明早期状态何时创建的时间戳作为选项。

RESTORE TABLE target_table TO VERSION AS OF <version>;
RESTORE TABLE target_table TO TIMESTAMP AS OF <timestamp>;

有关语法详细信息，请参阅 RESTORE。

在前面的示例中，该 RESTORE 命令会导致更新，这些更新已在读取表的版本 0 和 1 时被观察到。 如果流式处理查询当时正在读取此表，则这些文件将被视为新添加的数据，因此会再次对其进行处理。

还原指标

RESTORE 在操作完成后以单行数据帧的形式报告以下指标：

• table_size_after_restore：表还原后的大小。

• num_of_files_after_restore：还原后表中的文件数。

• num_removed_files：从表中删除（逻辑删除）的文件数。

• num_restored_files：由于回退而还原的文件数。

• removed_files_size：从表中删除的文件的总大小（以字节为单位）。

• restored_files_size：已还原的文件的总大小（以字节为单位）。

  

使用时间旅行的示例

• 为用户 111 修复对表的意外删除问题：

  INSERT INTO my_table
    SELECT * FROM my_table TIMESTAMP AS OF date_sub(current_date(), 1)
    WHERE userId = 111
  

• 修复表中意外发生的错误更新：

  MERGE INTO my_table target
    USING my_table TIMESTAMP AS OF date_sub(current_date(), 1) source
    ON source.userId = target.userId
    WHEN MATCHED THEN UPDATE SET *
  

• 查询在过去一周内增加的新客户的数量。

  SELECT
  (
    SELECT count(distinct userId)
    FROM my_table
  )
  -
  (
    SELECT count(distinct userId)
    FROM my_table TIMESTAMP AS OF date_sub(current_date(), 7)
  ) AS new_customers
  

如何在 Spark 会话中找到上次提交的版本？

在所有线程和所有表中，要获取当前 SparkSession 写入的最后一次提交的版本号，请查询 SQL 配置 spark.databricks.delta.lastCommitVersionInSession。

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

spark.conf.get("spark.databricks.delta.lastCommitVersionInSession")

Scala

spark.conf.get("spark.databricks.delta.lastCommitVersionInSession")

如果 SparkSession 未进行任何提交，则查询该键时将返回空值。