Databricks仓库遍历表提取最新Timestamp至Delta表的PySpark问题

需求是用PySpark从Databricks仓库指定表提取最新（max）timestamp，以overwrite方式写入存储旧timestamp的现有Delta表，且代码要适配任意数量的表。以下是尝试过程中遇到的问题及解决方法：

初始尝试代码 #

筛选目标表SQL #

%sql SHOW TABLES FROM database1 LIKE 'date_stamp'

提取Timestamp的Python代码 #

from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
df = sqlContext.sql("SELECT timestamp FROM table_date_stamp_source1")
df_filtered=df.filter(df.timestamp.max)

更新Delta表的代码 #

from pyspark.sql.functions import when
final_df = final_df.withColumn("timestamp_max", when(final_df.source == "table_data_stamp_source1" , final_df.timestamp_max == df_filtered.timestamp) \
      .otherwise(final_df.timestamp_max))

遍历多表时的报错代码 #

df_relevant_Tables=sqlContext.sql("SHOW TABLES FROM db1 LIKE '*date*' ")
df_relevant_Tables.select(df_relevant_Tables.columns[1])
for index, row in df_relevant_Tables.iterrows():
    df_name = row
    ...
    latest_date=df.select(max("db1.{df_name}.timestamp_column"))

报错信息 #

[UNRESOLVED_COLUMN.WITH_SUGGESTION] A column or function parameter with name `z` cannot be resolved. Did you mean one of the following? [`spark_catalog`.`db1`.`df_name`.`timestamp_column`];
'Project ['z]

问题原因与解决方法 #

核心错误点 #

1. DataFrame遍历方式错误：PySpark DataFrame不能用iterrows()（这是Pandas的方法），要改用collect()获取行数据。
2. 字符串格式化错误："db1.{df_name}.timestamp_column"的语法不对，Python里要用f-string或者format()，且row是Row对象，得提取表名字段（比如row.tableName）。
3. Max Timestamp提取逻辑错误：原代码df.filter(df.timestamp.max)完全错误，应该用agg(max("timestamp"))聚合获取最大值。

完整通用实现代码 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import max, lit

# Databricks中可直接用内置的spark对象，无需手动创建SparkSession
spark = SparkSession.builder.appName("UpdateDeltaMaxTimestamp").getOrCreate()

# 配置参数
SOURCE_DB = "db1"
TARGET_DELTA_TABLE = "your_target_delta_table"  # 存储各表最新timestamp的Delta表
TIMESTAMP_COL = "timestamp_column"  # 统一的timestamp列名，若各表列名不同需调整逻辑

# 获取所有符合条件的表
tables_df = spark.sql(f"SHOW TABLES FROM {SOURCE_DB} LIKE '*date*'")
# 提取表名列表
table_names = [row.tableName for row in tables_df.collect()]

# 构建各表最新timestamp的数据集
update_data = []
for table_name in table_names:
    # 读取表并计算max timestamp
    max_ts_df = spark.sql(f"SELECT MAX({TIMESTAMP_COL}) AS max_timestamp FROM {SOURCE_DB}.{table_name}")
    max_ts = max_ts_df.first()["max_timestamp"]
    # 存入表名和对应最新时间戳
    update_data.append((table_name, max_ts))

# 转换为Spark DataFrame，字段需和目标Delta表匹配
update_df = spark.createDataFrame(update_data, schema=["source_table", "timestamp_max"])

# 覆盖写入目标Delta表
update_df.write.format("delta").mode("overwrite").saveAsTable(TARGET_DELTA_TABLE)

补充说明 #

• 如果目标Delta表需要保留历史记录而非全量覆盖，可把mode("overwrite")改为mode("append")，但要额外处理重复表名的更新逻辑（比如用Delta的merge操作）。
• 若各表的timestamp列名不一致，可维护一个表名-列名映射字典，遍历时代入对应列名即可。
• Databricks环境中无需手动创建SQLContext，直接用内置的spark对象就行。

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