掌握 PySpark 的 groupBy，实现可扩展的数据聚合

在大数据世界里，我们很少直接面对原始数据。为了让数据更易于理解，对数据进行分组与聚合是一种常见且强大的操作。在使用 Apache Spark 与 Python 等分布式系统时，借助 PySpark 的 groupBy 函数，我们可以对分布式数据进行归集与汇总。它与 SQL 的 GROUP BY 子句功能相似，但专为高效处理海量数据的分布式数据处理而设计。

PySpark 的 groupBy 允许用户基于多个列对数据进行分区，随后可聚合为各种度量，如求和、均值等。为了高效实现这一点，PySpark 遵循“拆分-应用-合并”的范式：

1. 根据条件将数据拆分为若干组，
2. 对每个分组应用聚合或变换逻辑，
3. 将结果合并为新的 DataFrame。

PySpark 的 groupBy() 方法应用求和的示例

PySpark 采用惰性求值，这意味着像 groupBy 这样的操作只有在调用 show() 或 collect() 等 action 时才会真正计算。PySpark 会先构建一个初始的 有向无环图（DAG），并对其进行性能优化，有助于 Spark 在执行前优化查询计划。

如果您还没机会使用 PySpark，强烈建议先学习一下 Big Data Fundamentals with PySpark。

创建 PySpark DataFrame

首先，我们需要一个 PySpark DataFrame。如果您需要回顾各类 PySpark 命令，请查看这份很棒的 PySpark DataFrame 速查表。

我们将介绍一些关键步骤与命令，帮助您快速开始使用 PySpark。在开始之前，请确保您的 Python 环境已安装 PySpark、Java 和 Java JDK。如需帮助，请参考这篇 PySpark 入门指南。

启动 SparkSession

第一步，让您的 PySpark 跑起来！

from pyspark.sql import SparkSession

# Start a SparkSession
spark = SparkSession.builder \
    .appName("GroupByExample") \
    .getOrCreate()技术支持

用于 GroupBy 的示例 DataFrame

让我们构建一个示例 DataFrame 来动手实践。

from pyspark.sql import Row

data = [
    Row(department="Sales", employee="Alice", salary=5000),
    Row(department="Sales", employee="Bob", salary=4800),
    Row(department="HR", employee="Carol", salary=4000),
    Row(department="HR", employee="David", salary=3900),
    Row(department="IT", employee="Eve", salary=6000)
]

df = spark.createDataFrame(data)
df.show()技术支持

输出：

	+----------+--------+------+
	|department|employee|salary|
	+----------+--------+------+
	|     Sales|   Alice|  5000|
	|     Sales|     Bob|  4800|
	|        HR|   Carol|  4000|
	|        HR|   David|  3900|
	|        IT|     Eve|  6000|
	+----------+--------+------+技术支持

什么是 PySpark 的 GroupBy？

PySpark 的 groupBy 是一种转换操作，用于基于一个或多个列将数据拆分为若干组，随后可分别进行聚合或变换。

基本语法与示例

使用 groupBy 方法很简单；直接在目标 DataFrame 上调用即可。只要列适合用于分区，任何类型的列都可以使用。例如，您可能会避免使用浮点数。

grouped = df.groupBy("department")技术支持

这会创建一个 GroupedData 对象，它本身不是 DataFrame，但会告诉 Spark 如何开始对数据进行分区。有了这个 GroupedData 对象后，您可以应用 .count() 或 .sum() 等聚合以获得新的 DataFrame。

grouped.count().show()技术支持

参数与返回值

该方法唯一的参数是 *cols，可接收列名、列表达式、列序号（int）或列的列表。只要该列是您希望用于分区的，就可以在操作中使用。它始终返回一个 GroupedData 对象。

# grouping by column name like above
	df.groupBy("department")

# grouping by column expression
	df.groupBy(df.department)

#grouping by column ordinal
	df.groupBy(1)

#grouping by list of columns, you can mix the methods!
	df.groupBy(["department", 2])技术支持

对单列与多列进行 GroupBy

如上所示，您可以基于单列或多列进行分组。仅对一列分组适用于只关心单一维度（如部门或年份）的场景。多列分组则适合需要更多层次细节的情况，例如不同部门中的具体销售人员，或某年中的具体月份。

# grouping by single column
	df.groupBy("department").sum(“salary”).show()

#grouping by multiple columns
	df.groupBy(["department", 2]).sum(“salary”).show()技术支持

您可能已经注意到，groupBy 和 SQL 的 GROUP BY 语句非常相似。稍后我们也会展示在 PySpark 中使用 SQL 的方式。

聚合函数与技巧

PySpark 内置了丰富的聚合方法来处理分组数据。如果您熟悉 SQL，应该对许多函数（如 count()、sum()、avg() 等）并不陌生。可参考我们的 SQL 聚合函数指南进行回顾。

内置聚合函数

下面来看常见的内置聚合函数：

• count()：统计该分区中的记录数
• sum()：对数值求和
• avg()：对数值求平均
• min()：返回该分区中的最小值
• max()：返回该分区中的最大值

除了聚合函数，您可以使用 .alias() 方法为列重命名，使其更易理解。我们会在下面展示示例。

使用 agg() 进行多重聚合

与其分别执行每个聚合，不如使用 agg() 方法一次性执行多个聚合。每个聚合都会在 DataFrame 中生成一个新列。这也减少了多次调用 groupBy 的需求，从而提升性能。使用 agg() 时并不需要在同一列上聚合，您可以为每个聚合函数指定不同的列。

# After already starting your session
from pyspark.sql import functions as sf

df.groupBy("department").agg(
    sf.count("employee").alias("employee_count"), 
    sf.avg("salary").alias("avg_salary"),
    sf.max("salary").alias("max_salary")
).show()技术支持

输出：

+----------+--------------+----------+----------+
|department|employee_count|avg_salary|max_salary|
+----------+--------------+----------+----------+
|     Sales|             2|    4900.0|      5000|
|        HR|             2|    3950.0|      4000|
|        IT|             1|    6000.0|      6000|
+----------+--------------+----------+----------+技术支持

高级聚合模式

PySpark 的优势还在于可以使用更高级的聚合模式，例如数据透视（pivot）、分级汇总（rollup）以及数据立方（cube）。您也可以创建分组集（grouping sets）。

透视（Pivoting）

类似在 Excel 中创建数据透视表，您可以在数据中基于不同列进行透视。在此示例中，我们按 department 分组，并在 employee 列上进行透视，以查看每位员工的总薪资。

这意味着每一行展示部门，而每一列代表该部门的员工。若我们有按年份的数据，并希望查看各部门逐年表现，这种方式会非常有用。

df.groupBy("department").pivot("employee").sum("salary").show()技术支持

Rollup 与 Cube

两种非常强大的聚合方式是 rollup() 和 cube()。与普通的 groupby() 仅展示已有聚合结果不同，rollup() 和 cube() 都是层级结构，即会在更细粒度上进行聚合。

例如，rollup() 会按从左到右的顺序聚合，展示逐步迭代的每种可能排列。对每个部门，我们既会展示每个人员级别的聚合，也会展示该部门的汇总。

而 cube() 则会对所有聚合列的所有可能组合进行汇总，因此会得到每个部门单独、每个员工单独，以及两者任意组合的结果。

总结如下：

• Rollup 按列的顺序创建分层小计。
• Cube 为指定列的每种可能组合生成小计。

参见下方代码与输出示例：

rollup() 代码：

# using rollup
df.rollup("department", “employee”).sum("salary").show()技术支持

rollup() 输出：

+----------+--------+-----------+
|department|employee|sum(salary)|
+----------+--------+-----------+
|      NULL|    NULL|      23700|
|     Sales|   Alice|       5000|
|     Sales|    NULL|       9800|
|     Sales|     Bob|       4800|
|        HR|   Carol|       4000|
|        HR|    NULL|       7900|
|        HR|   David|       3900|
|        IT|     Eve|       6000|
|        IT|    NULL|       6000|
+----------+--------+-----------+技术支持

cube() 代码：

#using cube
df.cube("department", "employee").sum("salary").show()技术支持

cube() 输出：

+----------+--------+-----------+
|department|employee|sum(salary)|
+----------+--------+-----------+
|      NULL|   Alice|       5000|
|      NULL|    NULL|      23700|
|     Sales|   Alice|       5000|
|     Sales|    NULL|       9800|
|     Sales|     Bob|       4800|
|      NULL|     Bob|       4800|
|      NULL|   Carol|       4000|
|        HR|   Carol|       4000|
|        HR|    NULL|       7900|
|        HR|   David|       3900|
|      NULL|   David|       3900|
|        IT|     Eve|       6000|
|        IT|    NULL|       6000|
|      NULL|     Eve|       6000|
+----------+--------+-----------+技术支持

分组集（Grouping sets）

分组集允许您定义多个聚合层级。在本例中，我在部门与员工层级、仅部门层级，以及全局层级上进行聚合以获取总薪资。

您会看到 groupingSets() 方法的语法稍有不同。首先定义分组集列表 [(“department”, “employee”), (“department”, ), ()]，其中第一个为“部门+员工”，第二个仅为“部门”，最后一个空括号表示“全部”。

然后在这些集合内定义聚合所需的列。其余部分与使用 .agg() 的写法相同。

df.groupingSets([("department", "employee"), ("department",), ()], "department","employee").agg(sf.sum("salary")).sort("department","employee").show()技术支持

输出：

+----------+--------+-----------+
|department|employee|sum(salary)|
+----------+--------+-----------+
|      NULL|    NULL|      23700|
|        HR|    NULL|       7900|
|        HR|   Carol|       4000|
|        HR|   David|       3900|
|        IT|    NULL|       6000|
|        IT|     Eve|       6000|
|     Sales|    NULL|       9800|
|     Sales|   Alice|       5000|
|     Sales|     Bob|       4800|
+----------+--------+-----------+技术支持

自定义聚合函数

最后，我们可以使用 PySpark 的用户自定义函数（UDF） 来创建自定义聚合函数。有两种方式：udf 与 pandas_udf。二者都能让您自定义函数，各有优劣。

标准的 Spark udf 允许您创建 Spark 原生函数，这意味着语法和数据类型需原生兼容 Spark。虽然这限制了 UDF 能做的一些事情，但可以充分利用 Spark 的分布式计算能力，更适合大数据集。

from pyspark.sql.functions import udf
from pyspark.sql.types import IntegerType

@udf(returnType=IntegerType())
def bonus(salary):
    return int(salary * 0.1)

df.withColumn("bonus", bonus(df.salary)).show()技术支持

另一方面，pandas_udf 让您可以编写更“Pythonic”的自定义函数。这样您只需遵循 Pandas 的能力而非仅限于 Spark。

不过，这意味着无法利用 Spark 的分布式算力，而是依赖本地计算。它更适用于小到中等规模的数据集。

import pandas as pd
from pyspark.sql.functions import pandas_udf

@pandas_udf("double")
def salary_bonus(s: pd.Series) -> pd.Series:
    return s * 0.1

df.withColumn("bonus", salary_bonus(df.salary)).show()技术支持

总体上，只有在没有更好的方式进行聚合时，才考虑使用自定义函数。不要重复造轮子！

另外，如果某个操作可以进行向量化（如上面的函数），请优先选择函数的向量化版本，而不是使用聚合 UDF。想进一步了解 groupBy()，可阅读这篇更深入探讨 pandas 的拆分-应用-合并框架的文章。

筛选聚合后的数据

在 PySpark 中，您可以在分组之后基于聚合指标使用 filter() 方法来筛选分组。该方法中可提供 Python 或 SQL 表达式作为条件。

同样地，您也可以使用 where()，它是 filter() 的别名，效果相同。

您可以选择在聚合之前或之后进行筛选。聚合前筛选会限制参与聚合的数据，从而影响结果并可能提升性能。

例如，也许我们只想统计薪资高于某一阈值的员工数量，以找出“高薪”员工数。

#filter for high salaries
filter_df = df.filter(df.salary > 4000)

#aggregate and find the total of high salary employees
agg_filter_df = filter_df.groupBy("department").agg(sf.count("*").alias("high_salary_emp"))
agg_filter_df.show()技术支持

然而，在聚合之后进行筛选则不会影响原始聚合。例如，我们想统计每个部门的员工总数，但只查看规模超过某一阈值的部门。这会增加一层处理，但通常不会显著提高计算负载。

# creating a dataframe counting the number of employees
agg_df = df.groupBy("department").agg(sf.count("*").alias("num_employees"))

# filtering for departments where there is more than 1 employee
agg_df.where("num_employees > 1").show()技术支持

请合理选择筛选时机，因为它会影响准确性与最终结果。在聚合之前筛选可能会降低最终总量，而在聚合之后筛选可能会让过多数据保留在最终结果中。

PySpark groupBy 的性能优化策略

虽然 PySpark 会自动优化 groupBy 操作，但仍有一些策略可以进一步提升处理速度。尽量减少数据洗牌、缓解数据倾斜以及优化执行计划，都能提升 groupBy 的性能。

Shuffle 管理

groupBy 会引发 shuffle，即为了将相同键的数据聚在一起而在分区间重新分配数据。Shuffle 会消耗网络与磁盘资源，因此尽量减少或优化 shuffle 至关重要。以下是一些关键优化技巧：

• 合理使用 repartition()，您可以指定依据哪一列进行分区，从而降低数据查找范围
• 调整 spark.sql.shuffle.partitions。默认情况下，PySpark 使用 200 个 shuffle 分区。若数据量较小，可降低该数值。
• 启用 shuffle 压缩。压缩可减少网络与磁盘开销。

下面是一些改进 shuffle 管理的示例。

# reduce the number of shuffle partitions
spark.conf.set("spark.sql.shuffle.partitions", "64")  # Adjust based on cluster size

# make sure spark compresses the data
spark.conf.set("spark.shuffle.compress", "true") # compresses network transfer
spark.conf.set("spark.shuffle.spill.compress", "true") # compresses disk spillage

# repartitions the data prior to grouping to optimize aggregation
df.repartition("department").groupBy("department").sum("salary").show()技术支持

缓解数据倾斜

当某些键出现的频率显著高于其他键时，就会发生数据倾斜，导致分区负载不均、任务变慢，并让部分 worker 承受更大压力。我们可以通过加盐（salting）、减少倾斜连接，以及对小维表使用广播连接来缓解。

• 加盐：新增一个随机数列，强制数据在 worker 间更均匀地分布
• 通过重新分区来最小化倾斜，强制 Spark 使用不同列进行分区，可能带来更好的负载分布

# Salting example:

df = df.withColumn("salted_key", sf.rand()) #Create column of random
df = df.repartition(2, 'salted_key') # use this to repartition data
df.groupBy(sf.spark_partition_id()).count().show()技术支持

执行优化

PySpark 作业的逻辑与物理执行计划可以通过内置功能优化，从而提升 groupBy 聚合的性能。

• Catalyst 优化器：Spark 会自动重写低效计划；编写声明式转换（而非过程式循环）有助于优化器发挥作用。
• 缓存：当同一 groupBy 结果在流水线中多次复用时，缓存非常有用。
• 广播连接：将较小的 DataFrame 广播到内存中，而让大数据集被拆分和分区，以尽量减少集群的网络与磁盘开销。

grouped_df = df.groupBy("department").sum("salary").cache()
grouped_df.show() # you can cache your intermediate results

# use broadcasting (pseudocode here):
from pyspark.sql.functions import broadcast
df.join(broadcast(smaller_df), "department").show()技术支持

关于如何优化 PySpark 的操作，想了解更多底层机制，可阅读这篇关于 PySpark Joins 的文章。

与 RDD 操作的对比分析

由于 Catalyst 优化器带来的高层抽象与性能优势，DataFrame API 是大多数 PySpark 用户的推荐接口。但理解 RDD（弹性分布式数据集）层对于需要更强控制力或迁移遗留 Spark 代码的用户也很有价值。总体而言，使用 DataFrame 或 SQL API 通常能获得更好的性能。

RDD 是 PySpark 的核心组件，一切（包括 DataFrame API）都构建于其之上。它倾向于在内存中工作，并在处理流数据方面相较 DataFrame API 略有优势。然而，RDD 的数据是不可变的，一旦创建便无法修改。

本节将比较 DataFrame API 中的 groupBy 与 RDD API 中的等价操作，并分析其性能与适用场景。

首先，让我们将 DataFrame 转换为 RDD。

# Convert DataFrame to RDD of (key, value)
rdd = df[['department','salary']].rdd
rdd.collect()技术支持

现在我们可以在 RDD 上执行 groupByKey() 和 reduceByKey() 等操作。先来看 groupByKey()。

groupByKey() 会将相同键的所有值 shuffle 到同一个 executor，然后进行分组。由于这一机制，数据为找到目标键将进行大量 shuffle，可能耗时很长，也容易导致数据倾斜。

rdd.groupByKey().mapValues(list).collect()技术支持

相反，我们可以使用 reduceByKey()。该方法会先在每个分区内进行聚合，再在网络间传输数据，从而最小化 shuffle 数据量，因此速度更快。

因此，reduceByKey 明显优于 groupByKey，因为它显著减少了 shuffle 操作。

它会先在各分区聚合，再合并数据并继续聚合。对于 sum()、max() 等按键合并的聚合，这种方式最佳。

from operator import add

rdd.reduceByKey(add).collect()技术支持

DataFrame 与 RDD 聚合方法的性能优势

下表总结了各聚合方法的优势。

Big Data Fundamentals with PySpark 更详细地介绍了使用 RDD 编程，并说明了其如何构成 PySpark 的基石。

PySpark SQL 的 GROUP BY 查询

在 PySpark 中执行聚合的另一种方式是使用 SQL API 直接编写 SQL 语句。如果您更习惯写 SQL，这会是个不错的选择。

第一步是使用 DataFrame 的 createOrReplaceTempView() 方法创建一个临时视图。然后使用 spark.sql() 编写语句。

# Create a temporary view using the DataFrame
df.createOrReplaceTempView("employees")

# Write a SQL-like statement
spark.sql("""
    SELECT department, AVG(salary) AS avg_salary
    FROM employees
    GROUP BY department
""").show()技术支持

输出：

+----------+----------+
|department|avg_salary|
+----------+----------+
|     Sales|    4900.0|
|        HR|    3950.0|
|        IT|    6000.0|
+----------+----------+技术支持

如您所见，这与编写普通 SQL 查询一样简单。使用 SQL API 的主要优势是熟悉度更高、查询编写更轻松。DataFrame API 往往在数据类型一致性与优化方面更有优势。

您还能使用 Spark SQL API 不易实现的 DataFrame 方法。想了解更多 SQL 的 GROUP BY，可阅读这篇关于 SQL 中 GROUP BY 与 HAVING 的文章。

真实世界的应用

数据聚合在现实中有非常多的应用。事实上，几乎可以肯定，您需要以某种方式聚合数据，才能进行分析与共享。

下面我们提供一些不同聚合用例的示例。其中部分可能是伪代码，或不完全适用于我们的测试数据集，目的只是帮助您理解如何编写这些聚合。

商业智能

我们反复提到的一个主题是将 groupBy 用作分层分析的手段。设想我们有一列“revenue”，且关心每个部门的表现。我们可以按部门分组，并评估各部门的营收贡献。

CopyEdit
df.groupBy("department").agg(sum("revenue").alias("total_revenue")).show()技术支持

时间序列分析

聚合使您可以使用窗口函数。这些函数会查看一个滑动窗口内的数据，既可以是按行顺序，也可以是按时间顺序。

我们先使用 Window 对象定义窗口，指定 partitonBy 与 orderBy。然后在聚合函数的 over() 方法中使用该对象。例如，滚动平均可写为 sf.avg().over(window).show()。

设想我们想查看各部门的薪资滚动平均。我们会按 department 分组，并按一个新列 employeeId 排序。由于员工 ID 往往是顺序编号，我们可以观察平均薪资随时间的变化。

from pyspark.sql.window import Window

#Define the window we wish to partition and order by
windowSpec = Window.partitionBy("department").orderBy("employeeId")

#Perform the sf.avg function on the “salary” column over the windowSpec.
df.withColumn("rolling_avg", sf.avg("salary").over(windowSpec)).show()技术支持

这与在 SQL 中使用窗口函数的方式非常相似。

媒体分析

假设我们想了解媒体公司的多种用户指标，如总观看时长和唯一视频数。我们可以先对 user_id 列进行 groupBy()，再用 .agg() 搭配 sum()、countDistinct() 等不同函数计算不同指标。

df.groupBy("user_id").agg(
    sf.sum("watch_time").alias("total_watch"), #sum minutes watched
    sf.countDistinct("video_id").alias("unique_views") #count unique videos watched
).show()技术支持

PySpark GroupBy 的最佳实践与避坑

在 PySpark 中进行数据聚合时，容易踩到一些坑，导致性能下降与运行时间过长。遵循这些最佳实践有助于确保性能、正确性与可扩展性。

常见陷阱

以下是一些常见做法，会把简单查询变成长时间运行的程序。

1. 在 RDD 中过度使用 groupByKey()：
• 避免在 RDD 中使用 groupByKey()，因为其机制会让 PySpark 在分区间 shuffle 数据，导致大量网络与磁盘开销。
• 相反，优先使用 reduceByKey()，或坚持使用 DataFrame API。
2. 未处理数据倾斜：
• 当某个分组占主导（例如某部门拥有数百万条记录），该任务会成为瓶颈。
• 使用加盐或自定义分区，通过 repartition() 重新分配分区负载以缓解倾斜。
3. 忘记串联聚合函数：
• 将 sum() 放在一个 groupBy 调用里，然后在另一行对同一分组数据再次 count()，会打断 PySpark 的优化能力，导致低效的计划。
• PySpark 采用惰性求值模型，会在执行 action 前生成完整计划。将转换串联起来（如 groupBy().agg().filter()）可帮助 Catalyst 优化器高效规划。
4. 错误使用 UDF：
• Python UDF 会阻碍 Spark 对查询进行充分优化，因为 Catalyst 无法针对其进行优化。
• 如有可能，优先使用 Spark 内置函数或 pandas UDF，以获得更好性能。
5. 内存管理不当：
• 聚合操作代价高、占用内存大。当多次复用聚合结果时，可能导致内存耗尽。
• 监控内存使用情况；在复用分组数据时考虑使用 persist() 或 cache()。

优化与效率提升技巧

以下方法可帮助您优化 PySpark 聚合，让其稳定高效运行！

空值处理

• 对包含空值的列进行聚合，可能产生意料之外的结果。
• 在聚合前使用 na.fill() 或 na.drop()。

df.na.fill({"salary": 0}).groupBy("department").sum("salary").show()技术支持

避免过度 shuffle

• 如上所述，shuffle 会导致操作变慢。在分组前按逻辑对数据进行重新分区以尽量减小 shuffle 规模。
• 根据数据量调整 spark.sql.shuffle.partitions，寻找最适合您数据的分区数。

使用执行计划分析

• 使用 .explain() 监控查询计划，了解物理执行情况。
• 留意是否存在宽依赖 shuffle、广播提示以及低效扫描。

df.groupBy("department").sum("salary").explain(True)技术支持

输出：

	== Parsed Logical Plan ==
	'Aggregate ['department], ['department, unresolvedalias('sum(salary#600L))]
	+- LogicalRDD [department#598, employee#599, salary#600L], false

	== Analyzed Logical Plan ==
	department: string, sum(salary): bigint
	Aggregate [department#598], [department#598, sum(salary#600L) AS sum(salary)#679L]
	+- LogicalRDD [department#598, employee#599, salary#600L], false

	== Optimized Logical Plan ==
	Aggregate [department#598], [department#598, sum(salary#600L) AS sum(salary)#679L]
	+- Project [department#598, salary#600L]
	   +- LogicalRDD [department#598, employee#599, salary#600L], false

	== Physical Plan ==
	AdaptiveSparkPlan isFinalPlan=false
	+- HashAggregate(keys=[department#598], functions=[sum(salary#600L)], output=[department#598, sum(salary)#679L])
	   +- Exchange hashpartitioning(department#598, 200), ENSURE_REQUIREMENTS, [plan_id=1841]
	      +- HashAggregate(keys=[department#598], functions=[partial_sum(salary#600L)], output=[department#598, sum#681L])
	         +- Project [department#598, salary#600L]
	            +- Scan ExistingRDD[department#598,employee#599,salary#600L]
	技术支持

使用 Spark UI 监控性能

• 使用 Spark Web UI 跟踪 stage、task，识别运行缓慢的聚合或数据倾斜。
• 您可以在这门课程中了解更多 Spark Web UI 的内容：Introduction to Spark SQL in Python。

利用配置调优

Spark 提供了许多可针对大型 groupBy 工作负载进行调优的配置。下表列出了一些关键配置。

备注与高级实现细节

对于高级用户或处理超大规模数据集的用户，理解 groupBy 的内部行为至关重要。下面介绍一些更细致的工作原理。

惰性求值与执行计划

包括 groupBy 在内的所有 PySpark DataFrame 操作都采用惰性求值。即在触发 show() 或 collect() 等 action 之前不会进行实际计算。

这使得 Catalyst 优化器可以重排、合并或移除操作以提升性能。请充分利用这一点，将多个聚合与方法串联起来，让 Spark 生成更优的执行计划。

返回类型与命名冲突

使用 groupBy() 会返回 GroupedData 对象。执行 sum() 等聚合函数后会返回一个类似 DataFrame 的对象。使用 show() 可显示结果。

请记住 groupBy().agg() 会返回一个带有新列名的新 DataFrame。务必使用 alias() 方法添加别名，以避免命名冲突并让输出更清晰。如果不设置别名，可能引发列名冲突，或在后续连接操作中出现问题。

from pyspark.sql import functions as F

df.groupBy("department") \

  .agg(

      sum("salary").alias("total_salary"),

      count("*").alias("employee_count")

  )技术支持

聚合结果的格式

需要注意的是，PySpark 返回的聚合结果不保证行顺序。请使用 orderBy() 获取可预测的输出与一致的排序。

df.groupBy("department").sum("salary").orderBy("department").show()技术支持

特定版本行为与限制

不同版本的 PySpark 存在一些特性变化与限制。Catalyst 于 1.3 引入，并在 2.0 获得大幅改进。一些方法直到更高版本才引入，例如 groupingSets()。

下表罗列了 PySpark 若干较大的功能变化。请确保使用的 Spark 与 PySpark 版本能支持您所需的特性！

结论

PySpark 的 groupBy 是分布式环境中进行数据聚合的核心工具。无论是按区域汇总数据、计算平均指标，还是执行复杂的多层级分析，groupBy 都为处理大数据工作负载提供了可扩展且灵活的 API。

请牢记以下最佳实践，以尽量降低运行时间与成本：

• 尽可能使用内置函数
• 避免过度 shuffle 与数据倾斜等性能陷阱。
• 利用 Spark 的优化与分析工具微调大型作业。

随着 PySpark 的不断发展，我们可以期待更智能的优化、对复杂聚合的更强原生支持，以及与 Pandas 和 SQL 风格语法的更好集成，甚至对大规模模型有更好的支持。若您有兴趣进一步学习 PySpark，请参考 DataCamp 的以下资源：

• Big Data with PySpark 学习路径
• Spark 面试题前 20 条
• Apache Spark 教程：使用 PySpark 进行机器学习
• 使用 R 包 sparklyr 入门 Spark
• Groups：协作式数据科学学习中心