本教程详细介绍了如何在pyspark中对dataframe的所有列同时应用多个聚合函数（如`min`和`max`），并以行式结构（每行代表一个聚合结果）展示。通过结合使用`select`进行初步聚合、`cache`优化性能以及`unionbyname`进行结果重塑，实现了灵活且高效的数据分析，避免了直接`agg`函数无法满足特定输出格式的问题。
在PySpark进行数据分析时，一个常见的需求是对DataFrame中的所有或指定列应用多个聚合函数，例如同时计算每列的最小值和最大值。虽然DataFrame.agg()方法能够轻松实现多列多函数的聚合，但其默认输出是将所有聚合结果展平为单行，这往往无法满足将不同聚合类型（如最小值和最大值）作为独立行呈现的需求。
例如，直接使用df.agg(*exprs)的方式，其中exprs = [min(c).alias(c), max(c).alias(c) for c in df.columns]，会生成一个包含所有列的最小值和最大值，但这些值都将并列在同一行中，而不是我们期望的“一行是所有列的最小值，另一行是所有列的最大值”的结构。
为了实现这种行式输出的聚合结果，我们需要一种更为精细的策略，结合PySpark的select、cache和unionByName等操作。
解决方案：多阶段聚合与结果重塑
以下步骤将详细演示如何通过分阶段处理来达到目标输出格式：
- 初步聚合所有最小值和最大值： 首先，对DataFrame的所有列分别计算其最小值和最大值。这些聚合结果将暂时存储在一个新的DataFrame的单行中，其中每一列对应一个聚合值（例如，min_col1, max_col1, min_col2, max_col2等）。
- 缓存中间结果： 为了避免重复计算，对包含所有聚合值的中间DataFrame进行缓存。这在处理大型数据集时尤为重要。
- 重塑结果为行式结构： 将缓存的单行聚合结果拆分为多个DataFrame，每个DataFrame代表一种聚合类型（例如，一个DataFrame只包含所有列的最小值，另一个只包含所有列的最大值）。在拆分过程中，为每个DataFrame添加一个标识列（如agg_type），并统一列名，以便后续合并。
- 合并结果： 使用unionByName()方法将重塑后的DataFrame合并，最终得到我们期望的行式输出。
示例代码与详细解释
让我们通过一个具体的PySpark代码示例来演示上述过程：
import operator
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
# 初始化Spark会话
spark = SparkSession.builder.appName("MultiFunctionAggregate").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|
# +-----+----+----+-----+
# 1. 初步聚合所有最小值和最大值
# 构建min聚合表达式列表，并为结果列添加'min_'前缀
min_vals = [F.min(c).alias(f'min_{c}') for c in df.columns]
# 构建max聚合表达式列表，并为结果列添加'max_'前缀
max_vals = [F.max(c).alias(f'max_{c}') for c in df.columns]
# 使用select执行所有聚合，结果是一个单行DataFrame
df_agg_raw = df.select(min_vals + max_vals)
print("初步聚合结果 (单行):")
df_agg_raw.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|
# +-------+-------+-------+--------+-------+-------+-------+--------+
# 2. 缓存中间结果
# 缓存df_agg_raw以提高后续操作的性能
df_agg_raw.cache()
# 3. 重塑结果为行式结构
# 为最小值行构建选择表达式：添加'agg_type'列，并将min_前缀的列重命名回原始列名
min_cols = operator.add(
[F.lit('min').alias('agg_type')], # 添加一个字面量列，标识聚合类型为'min'
[F.col(f'min_{c}').alias(c) for c in df.columns] # 选取带有'min_'前缀的列，并将其别名改回原始列名
)
# 为最大值行构建选择表达式，原理同上
max_cols = operator.add(
[F.lit('max').alias('agg_type')], # 添加一个字面量列，标识聚合类型为'max'
[F.col(f'max_{c}').alias(c) for c in df.columns] # 选取带有'max_'前缀的列，并将其别名改回原始列名
)
# 从缓存的df_agg_raw中选择并重命名列，创建最小值DataFrame
min_df = df_agg_raw.select(min_cols)
# 从缓存的df_agg_raw中选择并重命名列，创建最大值DataFrame
max_df = df_agg_raw.select(max_cols)
print("重塑后的最小值DataFrame:")
min_df.show()
# +--------+-----+----+----+-----+
# |agg_type|col_1|col2|col3|col_4|
# +--------+-----+----+----+-----+
# | min| 2| 5| 18| 29|
# +--------+-----+----+----+-----+
print("重塑后的最大值DataFrame:")
max_df.show()
# +--------+-----+----+----+-----+
# |agg_type|col_1|col2|col3|col_4|
# +--------+-----+----+----+-----+
# | max| 8| 123| 26| 187|
# +--------+-----+----+----+-----+
# 4. 合并结果
# 使用unionByName合并两个DataFrame，确保按列名匹配
result = min_df.unionByName(max_df)
print("最终结果DataFrame:")
result.show()
# +--------+-----+----+----+-----+
# |agg_type|col_1|col2|col3|col_4|
# +--------+-----+----+----+-----+
# | min| 2| 5| 18| 29|
# | max| 8| 123| 26| 187|
# +--------+-----+----+----+-----+
# 停止Spark会话
spark.stop()注意事项与总结
- 列名管理： 在聚合阶段，通过alias()为聚合结果列添加前缀（如min_，max_）是关键，这有助于在后续重塑阶段清晰地识别和操作这些列。
- operator.add 的使用： 示例中operator.add用于连接两个列表，它等同于简单的列表拼接操作（list1 + list2）。
- F.lit()的作用： F.lit('min')或F.lit('max')用于创建一个字面量列，其值在所有行中都相同。这对于标识不同聚合类型至关重要。
- F.col()与alias()： 在重塑阶段，F.col(f'min_{c}').alias(c)的作用是选取带有特定前缀的列，并将其重命名回原始的列名，以保持最终结果的列名一致性。
- cache()的重要性： df_agg_raw.cache()在执行min_df和max_df的select操作之前，将中间聚合结果持久化到内存中。这避免了每次创建min_df和max_df时都重新计算原始DataFrame的聚合，显著提升了性能。
- unionByName()： unionByName()是合并具有相同列名但可能顺序不同的DataFrame的理想选择。它会根据列名进行匹配，而不是列的物理位置，从而增加了代码的健壮性。
- 扩展性： 这种方法不仅限于min和max，您可以轻松扩展到其他聚合函数（如avg, sum, count等），只需相应地修改聚合表达式和重塑逻辑即可。
通过上述方法，我们能够灵活地控制PySpark聚合结果的输出格式，满足将不同聚合类型以行式结构呈现的特定分析需求，同时兼顾了性能优化。
