Palantir Foundry PySpark非增量Transform空Schema错误排查求助

问题背景 #

在Palantir Foundry仓库的PySpark Transform中，原使用@transform incremental每月追加数据到历史_h表，现需新增逻辑：重跑某月数据时，删除历史表中对应reference_dt的数据并替换为新数据。改用普通@transform后出现未识别的空Schema错误，怀疑是历史_h表仅含一个月数据时的重写操作导致问题。

代码问题分析 #

原代码存在多处语法与逻辑问题，是空Schema错误的核心原因：

• 中文符号误用：代码中所有中文引号“”、多括号错误（如select([F.max(“reference_dt”)]].distinct()）会导致语法解析失败，进而引发Schema异常。
• DataFrame写入方式错误：Foundry中write_dataframe的正确调用格式为Output_h.write_dataframe(df, mode="overwrite")，原代码Output_h.write_dataframe.mode(“overwrite”)(output_new)不符合API规范。
• 不必要的distinct操作：F.max("reference_dt")本身返回唯一值，后续调用.distinct()属于冗余操作。
• 空DataFrame处理逻辑冗余：当output_prov2为空时，直接使用df.limit(0)即可保留Schema，无需额外获取schema变量。
• 未处理输入空数据场景：若my_input为空，collect()[0][0]会抛出索引越界异常，进而导致Schema无法识别。

修复后的代码 #

from pyspark.sql import functions as F

@transform(
    Output_h=Output(table_h),
    my_input=Input(table)
)
def my_compute_function(my_input, Output_h):
    df = my_input.dataframe()
    
    # 处理输入为空的情况，直接清空输出表
    if df.rdd.isEmpty():
        Output_h.write_dataframe(df.limit(0), mode="overwrite")
        return
    
    # 获取当前输入数据的最大reference_dt（即待替换的日期）
    distinct_dt = df.select(F.max("reference_dt")).collect()[0][0]
    
    # 读取历史表数据，过滤掉待替换的日期
    output_prov = Output_h.dataframe()
    output_prov2 = output_prov.filter(F.col("reference_dt") != distinct_dt)
    
    # 合并过滤后的历史数据与新数据（按列名匹配，避免列顺序问题）
    output_new = output_prov2.unionByName(df, allowMissingColumns=False)
    
    # 覆写输出表
    Output_h.write_dataframe(output_new, mode="overwrite")

关键优化说明 #

• 使用unionByName替代union：确保两个DataFrame按列名匹配合并，避免因列顺序不一致导致的Schema错误。
• 增加输入空数据处理：防止因输入无数据引发的索引越界与空Schema问题。
• 修正API调用格式：遵循Foundry的write_dataframe规范，避免语法错误。
• 移除冗余操作：删除不必要的distinct()调用，简化逻辑。

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