如何基于列列表实现Spark DataFrame的groupBy聚合操作？

当然支持通过列列表动态实现聚合操作，不用硬编码每一列的聚合逻辑。以下是几种实用的实现方式：

同一聚合函数应用到多列 #

如果需要对指定列列表统一使用某一种聚合函数（比如你的例子里的max），可以通过列表推导式动态生成聚合表达式，再用*解包传入agg()：

from pyspark.sql import functions as F

# 定义需要聚合的列列表（可动态生成或从配置读取）
cols_to_agg = ["a", "b"]
# 生成每个列对应的max聚合表达式，保持列名不变
agg_exprs = [F.max(col).alias(col) for col in cols_to_agg]
# 执行分组聚合
df = df.groupBy("x", "y").agg(*agg_exprs)

不同列应用不同聚合函数 #

如果需要给不同列指定不同的聚合函数，可以通过字典定义列与函数的映射，再生成对应的表达式：

from pyspark.sql import functions as F

# 定义列名与聚合函数的映射（支持Spark SQL函数或字符串函数名）
agg_specs = {
    "a": F.max,
    "b": F.min,
    "c": F.avg
}
# 动态生成聚合表达式
agg_exprs = [func(col).alias(col) for col, func in agg_specs.items()]
# 执行分组聚合
df = df.groupBy("x", "y").agg(*agg_exprs)

这种方式的优势在于列列表或聚合规则可以完全动态生成，比如从配置文件读取、根据业务逻辑动态筛选列等，彻底摆脱硬编码的限制。

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