为 Delta 表使用行跟踪

行跟踪允许 Delta Lake 跟踪 Delta Lake 表中的行级血缘。在 Delta Lake 表上启用行跟踪后，表会添加两个新的元数据字段

• 行 ID 为行提供在表中唯一的标识符。使用 MERGE 或 UPDATE 语句修改行时，行会保持相同的 ID。
• 行提交版本记录了行被修改时表的最后一个版本。使用 MERGE 或 UPDATE 语句修改行时，行会被分配一个新的版本。

注意

此功能在 Delta Lake 3.2.0 及更高版本中可用。在现有非空表上启用此功能在 Delta Lake 3.3.0 及更高版本中可用。

启用行跟踪

注意

创建时启用了行跟踪的表，会在创建时启用行跟踪 Delta Lake 表功能，并使用 Delta Lake 写入器版本 7。表协议版本无法降级，并且启用了行跟踪的表无法由不支持所有已启用的 Delta Lake 写入器协议表功能的 Delta Lake 客户端写入。请参阅Delta Lake 如何管理功能兼容性？。

您必须使用以下方法之一显式启用行跟踪

• 新表：在 CREATE TABLE 命令中设置表属性 delta.enableRowTracking = true。

SQL

-- Create an empty table
CREATE TABLE student (id INT, name STRING, age INT)
TBLPROPERTIES ('delta.enableRowTracking' = 'true');

-- Using a CTAS statement
CREATE TABLE course_new
TBLPROPERTIES ('delta.enableRowTracking' = 'true')
AS SELECT * FROM course_old;

-- Using a LIKE statement to copy configuration
CREATE TABLE graduate LIKE student;

-- Using a CLONE statement to copy configuration
CREATE TABLE graduate CLONE student;

• 现有表：从 Delta 3.3 及更高版本开始，在 ALTER TABLE 命令中设置表属性 'delta.enableRowTracking' = 'true'。

SQL

ALTER TABLE grade SET TBLPROPERTIES ('delta.enableRowTracking' = 'true');

• 所有新表：在 SET 命令中为当前会话设置配置 spark.databricks.delta.properties.defaults.enableRowTracking = true。

SQL

SET spark.databricks.delta.properties.defaults.enableRowTracking = true;

Python

spark.conf.set("spark.databricks.delta.properties.defaults.enableRowTracking", True)

Scala

spark.conf.set("spark.databricks.delta.properties.defaults.enableRowTracking", true)

注意

由于克隆 Delta Lake 表会创建单独的历史记录，因此克隆表上的行 ID 和行提交版本与原始表不匹配。

注意

在现有表上启用行跟踪将自动为表中所有现有行分配行 ID 和行提交版本。此过程可能会导致创建多个新版本的表，并且可能需要很长时间。

行跟踪存储

启用行跟踪可能会增加表的大小。Delta Lake 将行跟踪元数据字段存储在数据文件中的隐藏元数据列中。某些操作，例如仅插入操作，不使用这些隐藏列，而是使用 Delta Lake 日志中的元数据来跟踪行 ID 和行提交版本。数据重组操作（例如 OPTIMIZE 和 REORG）会导致使用隐藏元数据列跟踪行 ID 和行提交版本，即使它们以前是使用元数据存储的。

读取行跟踪元数据字段

行跟踪添加了以下可在读取表时访问的元数据字段

列名	类型	值
_metadata.row_id	长整型	行的唯一标识符。
_metadata.row_commit_version	长整型	行最后一次插入或更新时的表版本。

读取表时，行 ID 和行提交版本元数据字段不会自动包含在内。相反，这些元数据字段必须从所有 Apache Spark 表中都可用的隐藏 _metadata 列中手动选择。

SQL

SELECT _metadata.row_id, _metadata.row_commit_version, * FROM table_name;

Python

spark.read.table("table_name") \
  .select("_metadata.row_id", "_metadata.row_commit_version", "*")

Scala

spark.read.table("table_name")
  .select("_metadata.row_id", "_metadata.row_commit_version", "*")

禁用行跟踪

可以禁用行跟踪以减少元数据字段的存储开销。禁用行跟踪后，元数据字段仍然可用，但每次操作触及行时，所有行都会被分配一个新的 ID 和提交版本。

SQL

ALTER TABLE table_name SET TBLPROPERTIES (delta.enableRowTracking = false);

Python

spark.sql("ALTER TABLE table_name SET TBLPROPERTIES (delta.enableRowTracking = false)")

Scala

spark.sql("ALTER TABLE table_name SET TBLPROPERTIES (delta.enableRowTracking = false)")

注意

禁用行跟踪不会删除相应的表功能，也不会降级表协议版本。

限制

存在以下限制

• 读取更改数据流时，无法访问行 ID 和行提交版本元数据字段。
• 一旦将行跟踪功能添加到表中，就无法在不重新创建表的情况下将其删除。