如何在Delta Lake的table_changes中使用最新版本替代固定版本号

你可以通过以下两种方式获取Delta表的最新版本，再替换到table_changes的查询语句中：

方法1：通过Spark SQL查询版本历史 #

先查询表的版本历史提取最新版本号，再动态拼接SQL：

# 获取Delta表的最新版本号
latest_version_result = spark.sql("SELECT version FROM DESCRIBE HISTORY 'TableName' ORDER BY version DESC LIMIT 1").collect()
latest_version = latest_version_result[0][0]

# 用最新版本号查询变更记录
delta_df = spark.sql(f"SELECT * FROM table_changes('TableName', {latest_version}) ORDER BY _commit_timestamp")
display(delta_df)

方法2：使用Delta Lake Python API #

直接通过DeltaTable类获取最新版本，代码更简洁：

from delta.tables import DeltaTable

# 加载目标Delta表（支持表名或路径）
delta_table = DeltaTable.forName(spark, "TableName")  # 若为路径存储，改用DeltaTable.forPath(spark, "/path/to/table")
latest_version = delta_table.version

# 查询最新版本对应的变更记录
delta_df = spark.sql(f"SELECT * FROM table_changes('TableName', {latest_version}) ORDER BY _commit_timestamp")
display(delta_df)

注意事项 #

• 若表是通过Hive元数据注册的，优先使用forName方法；若直接存储在路径下，使用forPath。
• table_changes函数的第二个参数指定起始版本，这里用最新版本会返回该版本的变更记录；如果需要从更早版本到最新版本的所有变更，可以设置起始版本为0。

内容的提问来源于stack exchange，提问作者Rohit Kulkarni