如何读取Delta表最新版本的Parquet文件？并行读取逻辑失效求助

一、只读取最新版本的文件 #

你直接遍历根目录的方式会把所有历史版本的Parquet文件都列出来，必然导致逻辑混乱。正确的做法是通过Delta Lake官方API获取最新版本的有效文件列表，而非自行解析文件名。

具体实现步骤 #

1. 加载Delta表对象，获取最新版本元数据
2. 从元数据中提取当前有效的Parquet文件路径
3. 将文件列表拆分后分配给并行作业处理

示例代码：

from delta.tables import DeltaTable

# 加载目标Delta表
delta_table = DeltaTable.forPath(spark, "abfss://path/<table>")

# 获取最新版本的有效文件列表（仅包含当前版本在用的文件）
latest_files = delta_table.currentSnapshot().files()

# 将文件按每组20个拆分，对应10个并行作业
file_groups = [latest_files[i:i+20] for i in range(0, len(latest_files), 20)]

# 循环处理每组文件（替换为你的作业执行逻辑）
for group_idx, file_group in enumerate(file_groups):
    df = spark.read.parquet(*file_group)
    # 后续数据推送等处理逻辑...

currentSnapshot().files()只会返回当前最新版本实际使用的Parquet文件，历史版本的冗余文件不会被包含，从根源解决了读取混乱的问题。

二、配置Delta表仅保留最新版本 #

如果业务不需要历史数据，可以通过以下两种方式清理旧版本：

1. 设置自动保留策略 #

修改Delta表属性，让系统自动快速清理旧版本日志和文件：

-- 修改已有表的属性
ALTER TABLE delta.`abfss://path/<table>` SET TBLPROPERTIES (
  'delta.logRetentionDuration' = '0 hours',
  'delta.deletedFileRetentionDuration' = '0 hours'
)

注意：该设置会彻底丢失所有历史版本，无法回滚数据，需确认业务无历史数据需求。

2. 手动清理旧版本（VACUUM） #

一次性清理所有旧版本文件：

-- 临时关闭安全检查（默认不允许清理7天内的文件）
SET spark.databricks.delta.retentionDurationCheck.enabled = false;

-- 清理所有超过0小时的旧文件（仅保留最新版本）
VACUUM delta.`abfss://path/<table>` RETAIN 0 HOURS;

-- 清理完成后重新开启安全检查
SET spark.databricks.delta.retentionDurationCheck.enabled = true;

原代码失效的原因 #

Delta表的每个版本都会在_delta_log目录记录元数据，历史版本的Parquet文件不会自动删除。直接遍历根目录会把所有版本的part-xxx文件都列出来，导致你按编号筛选的逻辑完全失效。必须通过Delta API获取当前有效文件，而非自行解析目录和文件名。

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