本教程详细介绍了如何在pyspark dataframe中对多个列应用多个聚合函数（如`min`和`max`），并将结果以行式结构呈现。通过分步演示，我们展示了如何利用`select`进行初步聚合，并结合`unionbyname`技巧将聚合结果重塑为易于分析的行式格式，适用于需要定制化聚合报告的场景。
在PySpark数据处理中，我们经常需要对DataFrame的多个列执行聚合操作，例如计算每个列的最小值和最大值。虽然PySpark的agg函数能够方便地进行多列多函数聚合，但其默认输出是将所有聚合结果并列在一行中。然而，在某些分析场景下，我们可能需要将不同聚合函数的结果以行（row-wise）的形式展示，即每一行代表一个聚合函数（如最小值、最大值），而列则对应原始DataFrame的列。本教程将详细介绍如何实现这种定制化的行式聚合输出。
1. 问题背景与常见误区
假设我们有一个PySpark DataFrame，并希望计算其中所有数值列的最小值和最大值。一个常见的初步尝试可能是使用列表推导式结合agg函数：
from pyspark.sql import functions as F # 假设 df 是一个 PySpark DataFrame # exprs = [F.min(c).alias(c), F.max(c).alias(c) for c in df.columns] # df2 = df.agg(*exprs)
这种方法虽然可以计算出所有列的最小值和最大值，但其结果会是一个单行DataFrame，其中包含类似 min_col1, max_col1, min_col2, max_col2 等列。这与我们期望的“第一行是所有列的最小值，第二行是所有列的最大值”的行式输出格式不符。
2. 实现行式聚合输出的策略
为了实现行式聚合输出，我们需要采取一种分两步走的策略：
- 初步聚合所有函数的结果到单行DataFrame： 首先，我们将所有需要的聚合函数（例如，每个列的min和max）应用到DataFrame，生成一个包含所有聚合结果的单行DataFrame。
- 重塑DataFrame为行式输出： 接着，我们将这个单行DataFrame拆分成多个逻辑行，每行代表一个聚合函数的结果，并通过unionByName将它们合并起来。
3. 详细实现步骤
让我们通过一个具体的例子来演示这个过程。
3.1 准备示例数据
首先，创建一个示例PySpark DataFrame：
import operator
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
# 初始化 SparkSession
spark = SparkSession.builder.appName("PySparkMultiAggTutorial").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()
# +-----+----+----+-----+
# |col_1|col2|col3|col_4|
# +-----+----+----+-----+
# | 4| 123| 18| 29|
# | 8| 5| 26| 187|
# | 2| 97| 18| 29|
# +-----+----+----+-----+3.2 第一步：初步聚合所有函数的结果
我们首先为每个列生成min和max的聚合表达式，并使用df.select()来执行这些聚合。这里使用select而不是agg是因为select可以接受多个表达式作为参数，并直接创建新的列。
# 为每个列生成 min 和 max 聚合表达式
min_vals = [F.min(c).alias(f'min_{c}') for c in df.columns]
max_vals = [F.max(c).alias(f'max_{c}') for c in df.columns]
# 将所有聚合表达式合并，并使用 select 得到一个单行 DataFrame
# 注意：这里也可以使用 df.agg(*min_vals, *max_vals)，效果类似
df_aggregated_single_row = df.select(min_vals + max_vals)
print("初步聚合后的单行DataFrame:")
df_aggregated_single_row.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|
# +-------+------+-------+--------+-------+-------+-------+--------+注意事项： 如果df_aggregated_single_row后续会被多次使用，为了优化性能，建议对其进行cache()操作：df_aggregated_single_row.cache()。
3.3 第二步：重塑DataFrame为行式输出
现在我们有了包含所有聚合结果的单行DataFrame (df_aggregated_single_row)。接下来，我们需要将其重塑为期望的行式输出。这涉及到为每种聚合类型（如min和max）创建单独的DataFrame，并添加一个标识聚合类型的列，然后通过unionByName合并它们。
# 1. 创建 min 结果的 DataFrame
# - 添加 'agg_type' 列标识为 'min'
# - 重命名聚合列回原始列名
min_cols = operator.add(
[F.lit('min').alias('agg_type')], # 添加聚合类型标识列
[F.col(f'min_{c}').alias(c) for c in df.columns] # 选择并重命名 min_xxx 列
)
min_df = df_aggregated_single_row.select(min_cols)
# 2. 创建 max 结果的 DataFrame
# - 添加 'agg_type' 列标识为 'max'
# - 重命名聚合列回原始列名
max_cols = operator.add(
[F.lit('max').alias('agg_type')], # 添加聚合类型标识列
[F.col(f'max_{c}').alias(c) for c in df.columns] # 选择并重命名 max_xxx 列
)
max_df = df_aggregated_single_row.select(max_cols)
# 3. 使用 unionByName 合并 min_df 和 max_df
# unionByName 要求两个 DataFrame 具有相同的列名和类型
result_df = min_df.unionByName(max_df)
print("\n最终行式聚合结果:")
result_df.show()
# +--------+-----+----+----+-----+
# |agg_type|col_1|col2|col3|col_4|
# +--------+-----+----+----+-----+
# | min| 2| 5| 18| 29|
# | max| 8| 123| 26| 187|
# +--------+-----+----+----+-----+这里的operator.add用于连接两个列表，它与直接使用+的效果相同，例如 [F.lit('min').alias('agg_type')] + [F.col(f'min_{c}').alias(c) for c in df.columns]。
4. 总结与扩展
这种方法提供了一个灵活且强大的模式，用于在PySpark中实现复杂的行式聚合输出。
- 核心思想： 将多函数聚合分解为两个阶段：首先进行所有聚合生成单行结果，然后通过选择、重命名和unionByName操作将单行结果重塑为多行。
- 可扩展性： 这种模式可以轻松扩展到更多的聚合函数，例如平均值 (F.avg)、标准差 (F.stddev)、计数 (F.count) 等。只需为每个新的聚合函数重复“生成聚合表达式 -> 创建新的 DataFrame -> 与现有结果 unionByName”的步骤即可。
- 性能考量： 对于大型DataFrame，df_aggregated_single_row.cache() 是一个重要的优化点，可以避免重复计算。
- 通用性： 这种方法不仅限于min和max，任何可以表示为PySpark SQL函数的聚合都可以通过类似的方式处理。
通过掌握这种技巧，开发者可以更灵活地控制PySpark聚合结果的呈现方式，以满足各种数据分析和报告的需求。
