本教程详细介绍了如何在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 operatorfrom pyspark.sql import SparkSessionfrom pyspark.sql import functions as F# 初始化 SparkSessionspark = 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可以接受多个表达式作为参数，并直接创建新的列。
喵记多
喵记多 – 自带助理的 AI 笔记
27 查看详情 
# 为每个列生成 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 DataFrame多列多函数聚合与行式结果呈现的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/588173.html
微信扫一扫 
支付宝扫一扫 
