如何在Delta Lake历史表中编写version字段的WHERE子句？

我正在查询Delta Lake表的历史记录，执行以下命令后得到了包含version、operationParameters等字段的历史输出：

describe history '/mnt/lake/BASE/SQLClassification/cdcTest/dbo/cdcmergetest/1'

已将表路径赋值给变量：

saveloc = '/mnt/lake/BASE/SQLClassification/cdcTest/dbo/cdcmergetest/1'

我已经知道几种读取特定版本数据的方式：

• PySpark指定版本参数：

  df4 = spark.read.option("versionAsof", 3).load(saveloc)
  

• 通过路径后缀指定版本：

  df5 = spark.read.load("/mnt/lake/BASE/SQLClassification/cdcTest/dbo/cdcmergetest/1@v3")
  # 或用变量拼接路径
  df6 = spark.read.load(saveloc+"@v3")
  

• SQL语法：

  SELECT * FROM saveloc@v3
  

现在想请教：是否可以直接通过WHERE子句过滤version字段，比如执行以下语句来获取版本大于2的数据？

Select * From saveloc
where version > 2

答案 #

直接用WHERE version > 2这种写法是行不通的，因为Delta Lake表的业务数据本身并不包含version字段，这个字段属于表的元数据（仅存在于历史记录中），不是表数据列的一部分。

如果需要获取版本大于2的所有数据，有两种可行思路：

1. 筛选版本列表后逐个读取合并
   先查询表的历史记录，筛选出符合条件的版本号，再逐个读取对应版本的数据并合并（可以添加version字段标识数据所属版本）：

   from pyspark.sql.functions import lit
   
   # 读取历史记录并筛选出version>2的版本
   history_df = spark.sql(f"DESCRIBE HISTORY '{saveloc}'")
   valid_versions = history_df.filter("version > 2").select("version").rdd.flatMap(lambda x: x).collect()
   
   # 合并多个版本的数据
   combined_df = None
   for v in valid_versions:
       temp_df = spark.read.option("versionAsof", v).load(saveloc)
       temp_df = temp_df.withColumn("version", lit(v))
       if combined_df is None:
           combined_df = temp_df
       else:
           combined_df = combined_df.unionByName(temp_df)
   

2. 结合变更数据捕获（CDC）追踪版本变更
   如果你的Delta表启用了CDC功能，可以通过CDC日志来获取不同版本间的数据变更，这种方式更适合追踪数据的修改记录，而非直接合并多版本的全量数据。

需要注意：如果你的需求是获取最新版本中所有在版本2之后发生过变更的数据，这和合并多个版本的全量数据是不同场景，需要通过版本对比或CDC来实现。

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