PySpark中使用collect_set触发MISSING_GROUP_BY错误的问题排查

错误根源 #

你用的collect_set是聚合函数，这类函数必须配合GROUP BY子句指定分组维度，直接在withColumn里调用会触发MISSING_GROUP_BY错误——Spark不知道该按哪些列分组来聚合生成action集合。你之前尝试按cols分组报错，是因为cols里包含了要聚合的action列，分组键只能是那些非聚合的列。

修复步骤 #

1. 先处理所有非聚合列：my_key、id、active、dt_start_end这些都是可以直接生成的普通列，先完成这部分处理。
2. 确定分组键：把所有非聚合列作为分组依据，也就是["my_key","id","active","dt_start_end"]。
3. 用groupBy+agg替代withColumn生成聚合列：通过agg方法调用collect_set来生成action列，而不是在withColumn里直接用聚合函数。
4. 移除distinct：groupBy已经按分组键聚合，结果天然是唯一的，不需要再去重。

修改后的完整代码 #

cols = ["my_key","id","active","dt_start_end","action"]
group_cols = ["my_key","id","active","dt_start_end"]  # 仅保留非聚合列作为分组键

return (df
        .withColumn("my_key", lit(f"{my_key}"))
        .withColumn("id", lit(f"{id}"))
        .withColumn("active", lit(True))
        .withColumn("dt_start_end", self.get_start_end(self))
        .groupBy(*group_cols)
        .agg(
            collect_set(
                struct(
                    lit("source_key").alias("key"),
                    col("action").alias("value"),
                    col("benefit").alias("benefit"),
                    col("text").alias("text")
                )
            ).alias("action")
        )
        .select(*cols)
      )

def get_start_end(self):
    return struct(
        to_timestamp(col("dt_start")).alias("start"),
        to_timestamp(col("dt_end")).alias("end")
    )

# 若想保留方法复用，可直接在agg里调用，修改后如下：
# def get_action(self):
#     return collect_set(
#         struct(
#             lit("source_key").alias("key"),
#             col("action").alias("value"),
#             col("benefit").alias("benefit"),
#             col("text").alias("text")
#         )
#     )
# 调用时替换agg部分为：.agg(self.get_action().alias("action"))

额外说明 #

• dt_start_end是struct类型，Spark可以正常处理这类复杂类型作为分组键，无需额外转换。
• 聚合函数只能出现在agg或select配合groupBy的场景中，不能直接用在withColumn这类逐行处理的方法里。

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