使用表历史记录

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

检索表历史记录

通过运行 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()）。 此不一致是预期行为，不应用于验证写入。

Example

请考虑按 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 列映射中操作指标的集合。

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

什么是时间旅行？

时间旅行支持根据时间戳或表版本（如事务日志中记录）查询以前的表版本。 可以对应用程序使用“按时间顺序查看”，如下所述：

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

时间旅行语法

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

• 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。
• delta.deletedFileRetentionDuration = "interval <interval>"：确定由 VACUUM 用来删除当前表版本中不再引用的数据文件的阈值。 默认为 interval 7 days。

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

将表还原到早期状态

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

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

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

还原指标

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 未进行任何提交，则查询该键时将返回空值。