本教程详细阐述了如何在 pyspark dataframe 中对多列应用多个聚合函数（如 `min` 和 `max`），并将聚合结果以行式结构清晰地展示出来。针对 `df.agg` 默认列式输出的限制，文章提供了一种通过分步聚合、数据重塑及 `unionbyname` 操作实现目标行式输出的有效策略，并辅以代码示例和性能考量。
在 PySpark 中处理大规模数据时，我们经常需要对 DataFrame 中的多个列执行各种聚合操作，例如计算每列的最小值、最大值、平均值等。一个常见的需求是将这些不同聚合函数的结果以行而非列的形式展示，即每一行代表一个聚合类型（如“最小值”或“最大值”），而列则对应原始 DataFrame 的字段。然而，PySpark 的 DataFrame.agg() 方法通常会将所有聚合结果作为新列添加到一行中，这与我们期望的行式输出格式有所不同。本教程将详细介绍如何通过分步聚合、数据重塑和巧妙的合并操作，实现这种灵活的行式聚合结果展示。
PySpark DataFrame 多列多函数聚合的挑战
考虑一个包含多列数据的 PySpark DataFrame。如果我们需要计算所有列的最小值和最大值，并希望最终结果像这样：
+--------+-------+-------+-------+-------+ |agg_type| col_1 | col_2 | col_3 | col_4 | +--------+-------+-------+-------+-------+ | min| (min)| (min)| (min)| (min)| | max| (max)| (max)| (max)| (max)| +--------+-------+-------+-------+-------+
直接使用 df.agg(*exprs) 表达式，其中 exprs 包含 min(c).alias(c) 和 max(c).alias(c)，将无法直接得到上述行式结构。agg 操作会生成一个单行 DataFrame，其中包含 min_col1, max_col1, min_col2, max_col2 等列。为了达到目标行式输出，我们需要一种不同的策略。
解决方案：分步聚合与结果重构
实现所需行式输出的核心思想是：首先执行所有必要的聚合，然后将这些聚合结果重塑为我们想要的行式结构，最后通过合并操作将它们堆叠起来。
1. 执行所有聚合操作并生成中间结果
首先，我们需要为 DataFrame 中的每一列计算其最小值和最大值。我们可以通过列表推导式结合 pyspark.sql.functions 模块中的聚合函数来实现。
import operator
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
# 初始化 SparkSession
spark = SparkSession.builder.appName("MultiFunctionAggregation").getOrCreate()
# 示例数据
_data = [
(4, 123, 18, 29),
(8, 5, 26, 187),
(2, 97, 18, 29),
]
_schema = ['col_1', 'col2', 'col3', 'col_4']
df = spark.createDataFrame(_data, _schema)
print("原始 DataFrame:")
df.show()
# 为每列创建最小值和最大值的聚合表达式
min_vals_exprs = [F.min(c).alias(f'min_{c}') for c in df.columns]
max_vals_exprs = [F.max(c).alias(f'max_{c}') for c in df.columns]
# 将所有聚合表达式合并，并通过 select 执行
# 此时 df2 将是一个单行 DataFrame，包含所有 min_ 和 max_ 列
df2 = df.select(min_vals_exprs + max_vals_exprs)
print("中间聚合结果 (单行):")
df2.show()
# +-----+----+----+-----+------+-----+------+-------+
# |min_col_1|min_col2|min_col3|min_col_4|max_col_1|max_col2|max_col3|max_col_4|
# +-----+----+----+-----+------+-----+------+-------+
# | 2| 5| 18| 29| 8| 123| 26| 187|
# +-----+----+----+-----+------+-----+------+-------+在 df2 中，我们得到了一个包含所有列的最小值和最大值的单行 DataFrame。为了后续操作的效率，特别是当 df2 较大或后续会被多次引用时，建议对其进行缓存：df2.cache()。
2. 重塑数据以实现行式展示
接下来，我们需要将 df2 中的数据重塑为两行：一行用于最小值，另一行用于最大值。这需要为每种聚合类型创建一个独立的 DataFrame，并添加一个标识列。
# 缓存中间结果以提高性能
df2.cache()
# 准备最小值行的数据：添加 'agg_type' 列，并重命名聚合列
min_cols_selection = operator.add(
[F.lit('min').alias('agg_type')], # 添加聚合类型标识
[F.col(f'min_{c}').alias(c) for c in df.columns] # 重命名 min_col 为原列名
)
min_df = df2.select(min_cols_selection)
print("最小值 DataFrame:")
min_df.show()
# +--------+-----+----+----+-----+
# |agg_type|col_1|col2|col3|col_4|
# +--------+-----+----+----+-----+
# | min| 2| 5| 18| 29|
# +--------+-----+----+----+-----+
# 准备最大值行的数据：同样添加 'agg_type' 列并重命名
max_cols_selection = operator.add(
[F.lit('max').alias('agg_type')], # 添加聚合类型标识
[F.col(f'max_{c}').alias(c) for c in df.columns] # 重命名 max_col 为原列名
)
max_df = df2.select(max_cols_selection)
print("最大值 DataFrame:")
max_df.show()
# +--------+-----+----+----+-----+
# |agg_type|col_1|col2|col3|col_4|
# +--------+-----+----+----+-----+
# | max| 8| 123| 26| 187|
# +--------+-----+----+----+-----+通过 F.lit() 创建一个常量列 agg_type，并使用列表推导式和 F.col().alias() 将 min_col_N 和 max_col_N 列重命名回原始的列名 col_N，这样 min_df 和 max_df 就拥有了相同的结构和列名。
3. 合并重塑后的结果
最后一步是将 min_df 和 max_df 垂直合并，形成最终的行式聚合结果。这里我们使用 unionByName() 方法，它能够根据列名进行合并，确保即使列顺序不同也能正确匹配。
# 使用 unionByName 合并最小值和最大值 DataFrame
result = min_df.unionByName(max_df)
print("最终行式聚合结果:")
result.show()
# +--------+-----+----+----+-----+
# |agg_type|col_1|col2|col3|col_4|
# +--------+-----+----+----+-----+
# | min| 2| 5| 18| 29|
# | max| 8| 123| 26| 187|
# +--------+-----+----+----+-----+
# 停止 SparkSession
spark.stop()完整代码示例
将上述所有步骤整合，得到一个完整的实现：
import operator
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
def aggregate_multiple_functions_row_wise(dataframe, functions_map):
"""
对 PySpark DataFrame 的所有列应用多个聚合函数，并以行式结构展示结果。
Args:
dataframe (DataFrame): 输入的 PySpark DataFrame。
functions_map (dict): 字典，键为聚合类型字符串（如 'min', 'max'），
值为对应的 PySpark 聚合函数（如 F.min, F.max）。
Returns:
DataFrame: 包含行式聚合结果的 DataFrame。
"""
all_aggs_exprs = []
for agg_type, func in functions_map.items():
all_aggs_exprs.extend([func(c).alias(f'{agg_type}_{c}') for c in dataframe.columns])
# 1. 执行所有聚合操作，生成单行中间结果
intermediate_df = dataframe.select(all_aggs_exprs)
intermediate_df.cache() # 缓存中间结果以提高性能
result_dfs = []
for agg_type in functions_map.keys():
# 2. 为每种聚合类型重塑数据
selection_cols = operator.add(
[F.lit(agg_type).alias('agg_type')],
[F.col(f'{agg_type}_{c}').alias(c) for c in dataframe.columns]
)
agg_df = intermediate_df.select(selection_cols)
result_dfs.append(agg_df)
# 3. 合并所有重塑后的结果
if not result_dfs:
return spark.createDataFrame([], schema=['agg_type'] + dataframe.columns)
final_result = result_dfs[0]
for i in range(1, len(result_dfs)):
final_result = final_result.unionByName(result_dfs[i])
intermediate_df.unpersist() # 释放缓存
return final_result
if __name__ == "__main__":
spark = SparkSession.builder.appName("MultiFunctionAggregationTutorial").getOrCreate()
_data = [
(4, 123, 18, 29),
(8, 5, 26, 187),
(2, 97, 18, 29),
]
_schema = ['col_1', 'col2', 'col3', 'col_4']
df = spark.createDataFrame(_data, _schema)
print("原始 DataFrame:")
df.show()
# 定义要应用的聚合函数
functions_to_apply = {
'min': F.min,
'max': F.max,
'avg': F.avg,
'sum': F.sum
}
# 调用函数获取行式聚合结果
final_agg_df = aggregate_multiple_functions_row_wise(df, functions_to_apply)
print("最终行式聚合结果:")
final_agg_df.show()
spark.stop()运行上述代码，你将看到一个包含 min、max、avg、sum 四种聚合类型，每种类型一行，且列名与原始 DataFrame 保持一致的整洁输出。
注意事项与性能考量
- cache() 的使用: 在 df2（中间聚合结果）上调用 cache() 是一个重要的性能优化措施。由于 min_df 和 max_df 乃至更多的聚合类型 DataFrame 都是从 df2 派生出来的，缓存 df2 可以避免 Spark 在每次 select 操作时都重新计算原始 DataFrame 的聚合结果。在不再需要 intermediate_df 时，使用 unpersist() 释放缓存是一个好习惯。
- 列命名策略: 在初始聚合时，使用 f'{agg_type}_{c}' 这样的命名约定（例如 min_col1，max_col1）非常关键。这使得后续重塑阶段能够清晰地识别并提取出特定聚合类型的值。
- unionByName() 的重要性: unionByName() 方法是实现不同聚合类型 DataFrame 合并的关键。它确保了即使在不同子 DataFrame 中列的顺序可能有所不同，也能根据列名进行正确的匹配和合并，避免了因列顺序不一致导致的错误。
- 扩展性: 这种方法具有很好的扩展性。如果需要添加更多的聚合函数（如 F.avg()、F.stddev() 等），只需在 functions_map 中添加对应的键值对即可，代码结构无需大的改动。
- 处理大量列或函数: 尽管此方法有效，但如果 DataFrame 包含极大量的列或需要应用非常多的聚合函数，生成的中间 DataFrame (df2) 将会有大量的列。这可能会对 Spark 的执行计划和内存使用造成一定压力。在极端情况下，可能需要考虑将聚合分批处理或探索其他更复杂的聚合策略（例如，如果聚合函数可以分组，则考虑使用 groupBy().pivot().agg()，但这通常用于不同维度而非不同聚合类型）。
- 与 agg().pivot() 的区别: agg().pivot() 通常用于将一个分类列的不同值转换为多列，并对这些新列进行聚合。而本教程的需求是将同一个数据列上的不同聚合函数结果转换为多行，两者应用场景有所不同。
总结
本教程提供了一种在 PySpark 中对 DataFrame 多列应用多函数聚合，并以行式结构展示结果的有效方法。通过分步执行聚合、巧妙重塑数据以及利用 unionByName() 进行合并，我们能够克服 df.agg() 默认列式输出的限制，生成更符合分析和报告需求的行式聚合视图。掌握这种技术，将有助于你在 PySpark 数据处理中实现更灵活和强大的数据分析能力。
