PySpark多列聚合与行处理技巧详解

目前golang学习网上已经有很多关于文章的文章了，自己在初次阅读这些文章中，也见识到了很多学习思路；那么本文《PySpark多列聚合与行式结果处理技巧》，也希望能帮助到大家，如果阅读完后真的对你学习文章有帮助，欢迎动动手指，评论留言并分享~

本教程详细介绍了如何在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. 实现行式聚合输出的策略

为了实现行式聚合输出，我们需要采取一种分两步走的策略：

1. 初步聚合所有函数的结果到单行DataFrame： 首先，我们将所有需要的聚合函数（例如，每个列的min和max）应用到DataFrame，生成一个包含所有聚合结果的单行DataFrame。
2. 重塑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聚合结果的呈现方式，以满足各种数据分析和报告的需求。

本篇关于《PySpark多列聚合与行处理技巧详解》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于文章的相关知识，请关注golang学习网公众号！