PySpark中用withColumn替代select时出现分组表达式为空错误排查

原可运行代码（使用select） #

def dev_prev_month(cleaned):
    df = cleaned
    
    df = df.select(
    F.coalesce(
        _sum(
            F.when(
                (F.col("ORDERS_VIA_ARTICLE") > 0) &
                (
                    (F.col("ORDER_SUCCESS_URL") != "%16237890%") &
                    (F.col("ORDER_SUCCESS_URL") != "%30427132%") &
                    (F.col("ORDER_SUCCESS_URL") != "%242518801%") |
                    (F.col("ORDER_SUCCESS_URL").isNull())
                ),
                F.col("ORDERS_VIA_ARTICLE")
            ).otherwise(F.lit(0))
        ),
        F.lit(0)
    ).alias("report_sum_orders_via_article")
    )
    
    return df

改写后的withColumn代码（运行报错） #

def dev_prev_month(clean_joined_traffic_data):
    df = clean_joined_traffic_data
    df = df.withColumn(
        "report_sum_orders_via_article",_sum(
                F.when(
                    (F.col("ORDERS_VIA_ARTICLE") > 0) &
                    (
                        (F.col("ORDER_SUCCESS_URL") != "%16237890%") &
                        (F.col("ORDER_SUCCESS_URL") != "%30427132%") &
                        (F.col("ORDER_SUCCESS_URL") != "%242518801%") |
                        (F.col("ORDER_SUCCESS_URL").isNull())
                    ),
                    F.col("ORDERS_VIA_ARTICLE")
                ).otherwise(F.lit(0)))
        )
   
    return df

报错信息 #

pyspark.sql.utils.AnalysisException: grouping expressions sequence is empty, and '!ri.foundry.main.transaction.123-123:ri.foundry.main.transaction.xxxx:master.ORDERS' is not an aggregate function.

问题原因与解决方法 #

核心原因 #

原select代码里的_sum（应为F.sum）属于全局聚合，会将整个DataFrame聚合为一行结果；但withColumn是在原有每行数据基础上新增列，PySpark不允许在withColumn中直接使用未分组的聚合函数——聚合操作必须搭配groupBy，或明确为全局聚合，但withColumn无法直接新增全局聚合列（会与原有行结构冲突）。另外原代码的coalesce用于处理聚合结果为null的情况（无符合条件行时sum返回null，替换为0），这部分逻辑也需保留。

正确实现方式 #

方案1：保留原DataFrame所有行，新增全局聚合列

如果需要保留原表所有行，同时添加全局聚合结果：

from pyspark.sql import functions as F

def dev_prev_month(clean_joined_traffic_data):
    df = clean_joined_traffic_data
    # 计算全局聚合值
    agg_df = df.select(
        F.coalesce(
            F.sum(
                F.when(
                    (F.col("ORDERS_VIA_ARTICLE") > 0) &
                    (
                        ((F.col("ORDER_SUCCESS_URL") != "%16237890%") &
                         (F.col("ORDER_SUCCESS_URL") != "%30427132%") &
                         (F.col("ORDER_SUCCESS_URL") != "%242518801%")) |
                        (F.col("ORDER_SUCCESS_URL").isNull())
                    ),
                    F.col("ORDERS_VIA_ARTICLE")
                ).otherwise(F.lit(0))
            ),
            F.lit(0)
        ).alias("report_sum_orders_via_article")
    )
    # 广播聚合结果后关联回原表
    df = df.crossJoin(F.broadcast(agg_df))
    return df

方案2：仅保留聚合后的单行结果

如果不需要原表行数据，直接用聚合逻辑即可（本质和原select逻辑一致）：

from pyspark.sql import functions as F

def dev_prev_month(clean_joined_traffic_data):
    df = clean_joined_traffic_data
    df = df.select(
        F.coalesce(
            F.sum(
                F.when(
                    (F.col("ORDERS_VIA_ARTICLE") > 0) &
                    (
                        ((F.col("ORDER_SUCCESS_URL") != "%16237890%") &
                         (F.col("ORDER_SUCCESS_URL") != "%30427132%") &
                         (F.col("ORDER_SUCCESS_URL") != "%242518801%")) |
                        (F.col("ORDER_SUCCESS_URL").isNull())
                    ),
                    F.col("ORDERS_VIA_ARTICLE")
                ).otherwise(F.lit(0))
            ),
            F.lit(0)
        ).alias("report_sum_orders_via_article")
    )
    return df

额外注意点 #

• 建议将_sum替换为F.sum，明确导入pyspark.sql.functions避免混淆。
• 条件逻辑中建议给(三个URL不等于条件) | (URL为空)整体加括号，虽然PySpark中&优先级高于|，原逻辑解析正确，但加括号可提升可读性，避免后续修改时出错。

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