Spark自定义函数调用报错：TypeError: 'Column' object is not callable

一、lead_function 问题排查与修正 #

核心问题： #

• 函数内部错误引用了外部变量 new_df，而非传入的参数 df，违背了函数封装原则，同时会引发变量作用域相关的错误，这也是触发 TypeError: 'Column' object is not callable 的关键原因之一。
• 虽然调用时传入 F.col() 生成的Column对象本身合法，但函数逻辑依赖外部未传入的变量，导致执行异常。

修正后的代码： #

from pyspark.sql import Window
import pyspark.sql.functions as F

def lead_function(df, id_col, subcatid_col, start_date_col, price_col):  
    window_spec_st = Window().partitionBy(id_col, subcatid_col).orderBy(start_date_col)
    # 使用传入的df参数，而非外部new_df，确保函数封装性
    return df.select("record_id", "subcat_id", "start_date", price_col) \
             .withColumn("lead", F.lead(price_col, 1).over(window_spec_st))

# 调用时可直接传入列名字符串，更简洁（也支持F.col形式）
change_amount_df = lead_function(new_df, "id", "subcat_id", "start_date", "price")

二、price_vs_lead 问题排查与修正 #

核心问题： #

1. API逻辑混淆：F.when() 返回的是Column对象，而groupBy()是DataFrame的专属方法，两者无法链式调用——你试图在列操作后直接执行DataFrame聚合，完全不符合Spark的API设计逻辑。
2. 语法错误：agg() 方法内部第二个count前多了一个冗余的点，导致语法解析失败；同时直接用字符串别名changes_count_alias做条件判断，必须通过F.col(changes_count_alias)引用对应列。
3. 函数设计错误：该函数试图同时完成列计算和DataFrame聚合，这两个操作属于不同层级，应当拆分开执行。

修正后的实现方案： #

方案1：拆分函数为列计算与聚合两步

# 生成变化标记列的函数
def get_change_flag(price, lead, alias="changes_count"):
    return F.when(price > lead, 1) \
            .when(lead > price, 2) \
            .otherwise(F.lit(0)).alias(alias)

# 执行聚合统计的函数
def calculate_price_changes(df, id_col, subcatid_col, flag_col="changes_count"):
    return df.groupBy(id_col, subcatid_col) \
             .agg(
                 F.count(F.when(F.col(flag_col) == 1, 1)).alias("increase"),
                 F.count(F.when(F.col(flag_col) == 2, 1)).alias("decrease")
             )

# 使用示例
temp_df = change_amount_df.select("*", get_change_flag(F.col("price"), F.col("lead")))
result_df = calculate_price_changes(temp_df, "id", "subcat_id")

方案2：封装为单一DataFrame操作函数

def price_vs_lead(df, price_col, lead_col, id_col, subcatid_col, changes_count_alias="changes_count"):
    # 先添加变化标记列
    df_with_flag = df.withColumn(
        changes_count_alias,
        F.when(F.col(price_col) > F.col(lead_col), 1)
        .when(F.col(lead_col) > F.col(price_col), 2)
        .otherwise(F.lit(0))
    )
    # 再执行聚合操作
    return df_with_flag.groupBy(id_col, subcatid_col) \
                       .agg(
                           F.count(F.when(F.col(changes_count_alias) == 1, 1)).alias("increase"),
                           F.count(F.when(F.col(changes_count_alias) == 2, 1)).alias("decrease")
                       )

# 调用示例
result_df = price_vs_lead(change_amount_df, "price", "lead", "id", "subcat_id")

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