使用表历史记录

对于 Apache Iceberg 和 Delta Lake 表，修改表的每个操作都会创建新的表版本。 使用历史记录信息来审核操作、回滚表或查询特定时间点的表（使用按时间顺序查看）。

检索表历史记录

DESCRIBE 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。

有关 Scala、Java和Python语法详细信息，请参阅 Delta Lake API 文档。

目录资源管理器 在“ 历史记录 ”选项卡上直观显示表历史记录。

历史模式

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 列分区的表。 创建表时， partitionBy 将填充：

df.write.format("delta") \
  .partitionBy("date") \
  .saveAsTable("sales_data")

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

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

将数据追加到此表时， partitionBy 会显示一个空数组：

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

-- Timestamp version
SELECT * FROM people10m@20190101000000000
-- Version number
SELECT * FROM people10m@v123

Python

# Timestamp version
spark.read.table("people10m@20190101000000000")
# Version number
spark.read.table("people10m@v123")

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

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

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

若要增加表的数据保留阈值，必须配置下表属性，并将其替换为<format>：deltaiceberg

• <format>.logRetentionDuration = "interval <interval>"：控制表的历史记录的保留时间长度。 默认值为 interval 30 days。
  • 在 Databricks Runtime 18.0 及更高版本中， logRetentionDuration 必须大于或等于 deletedFileRetentionDuration。 对于 Unity 目录托管表，这适用于 Databricks Runtime 12.2 及更高版本。
• <format>.deletedFileRetentionDuration = "interval <interval>"：确定由 VACUUM 用来删除当前表版本中不再引用的数据文件的阈值。 默认值为 interval 7 days。

例如，若要访问 30 天的历史数据，请设置 delta.deletedFileRetentionDuration = "interval 30 days"，这与 delta.logRetentionDuration 的默认设置一致。

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

时间旅行示例

若要修复用户 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

事务日志检查点

事务日志以 JSON 文件的形式记录表版本，这些文件位于事务日志目录中，并与表数据并存。

为了优化检查点查询，将各表版本聚合成 Parquet 检查点文件，这样就无需读取表历史记录中的所有 JSON 版本，从而提升性能。 用户无需直接与检查点交互。

Azure Databricks 针对数据大小和工作负荷优化检查点操作频率。 检查点频率可能会更改，恕不另行通知。

将表还原到早期状态

使用 RESTORE 命令可将表还原到先前的版本或时间戳，包括以下场景：

• 可以再次还原已经还原的表。
• 可以还原克隆的表。

请考虑以下要求：

• 若要还原表，必须具有 MODIFY 该表的权限。
• 在数据文件被手动删除或由 VACUUM 删除后，你将无法将表还原到引用这些文件的较早版本。 如果将 spark.sql.files.ignoreMissingFiles 设置为 true，则仍可部分还原为此版本。
• 若要按时间戳还原，请使用格式 yyyy-MM-dd HH:mm:ss 或 yyyy-MM-dd。

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

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

流式处理行为

还原是一项数据更改操作，可能会导致下游工作负荷重复数据。 命令添加的 RESTORE 日志条目包含 dataChange 设置为 true。

对于下游工作负荷（例如处理表更新的 结构化流式处理 作业），还原操作添加的数据更改日志条目被视为新的数据更新，处理它们可能会导致重复数据。

例如：

在前面的示例中，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：已还原的文件的总大小（以字节为单位）。

  

查找最后一个提交版本

在所有线程和所有表中，要获取当前 SparkSession 写入的最后一次提交的版本号，请查询 SQL 配置 spark.databricks.<format>.lastCommitVersionInSession。 根据表格的格式，将 <format> 替换为 delta 或 iceberg。

例如：

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Python

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

Scala

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

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