PySpark 分组后再对组内排序

在本文中，我们将介绍如何在 PySpark 中使用 groupBy 函数对数据进行分组，并在每个组内对数据进行排序的方法。

什么是 PySpark？

PySpark 是 Apache Spark 在 Python API 上的开源分布式计算系统。它提供了一种强大的处理大规模数据的方式，并结合了 Python 的简易性和 Spark 的高性能。

PySpark 分组

在 PySpark 中，我们可以使用 groupBy 函数对数据进行分组。groupBy 函数接收一个或多个列作为参数，并将数据按照这些列的值进行分组。接下来，我们将通过一个示例来说明如何使用 groupBy 函数。

假设我们有一个包含学生姓名、科目和成绩的数据集，如下所示：

+-------+-------+-------+
|  Name | Subject | Score |
+-------+-------+-------+
| Alice |   Math  |   90  |
| Alice | Physics |   85  |
|  Bob  |   Math  |   80  |
|  Bob  | Physics |   95  |
| Carol |   Math  |   70  |
| Carol | Physics |   75  |
+-------+-------+-------+

现在，我们想要按照学生姓名对数据进行分组。我们可以使用以下代码来实现：

from pyspark.sql import SparkSession

# 创建 SparkSession
spark = SparkSession.builder.getOrCreate()

# 创建 DataFrame
data = [("Alice", "Math", 90),
        ("Alice", "Physics", 85),
        ("Bob", "Math", 80),
        ("Bob", "Physics", 95),
        ("Carol", "Math", 70),
        ("Carol", "Physics", 75)]

df = spark.createDataFrame(data, ["Name", "Subject", "Score"])

# 对数据进行分组
grouped_df = df.groupBy("Name")

上述代码中，我们首先创建了一个 SparkSession 对象，并使用 createDataFrame 方法创建了一个 DataFrame 对象。然后，我们调用 groupBy 方法，并将需要分组的列名作为参数传入。

分组后排序

在 PySpark 中，可以使用 orderBy 或 sort 方法对数据进行排序。这两个方法都接收一个或多个列名作为参数，并对数据按照这些列进行排序。接下来，我们将通过一个示例来说明如何在分组后对数据进行排序。

假设我们想要对上文中分组后的数据按照成绩进行降序排序。我们可以使用以下代码实现：

from pyspark.sql import functions as F

# 对分组后的数据进行排序
sorted_df = grouped_df.sort(F.desc("Score"))

上述代码中，我们使用 sort 方法对分组后的 DataFrame 进行排序。通过 F.desc 方法，我们可以指定按照列名 “Score” 进行降序排序。

除了 sort 方法，我们还可以使用 orderBy 方法对数据进行排序。下面是一个使用 orderBy 方法的示例：

# 对分组后的数据进行排序
sorted_df = grouped_df.orderBy(F.desc("Score"))

使用 orderBy 和 sort 方法效果是相同的，只是方法名称不同。

示例

我们将继续以之前的学生成绩数据为例，演示如何在 PySpark 中进行分组后的排序。

假设我们想要按照学生姓名对数据进行分组，并在每个组内按照成绩进行降序排序。我们可以使用以下代码实现：

from pyspark.sql import SparkSession, functions as F

# 创建 SparkSession
spark = SparkSession.builder.getOrCreate()

# 创建 DataFrame
data = [("Alice", "Math", 90),
        ("Alice", "Physics", 85),
        ("Bob", "Math", 80),
        ("Bob", "Physics", 95),
        ("Carol", "Math", 70),
        ("Carol", "Physics", 75)]

df = spark.createDataFrame(data, ["Name", "Subject", "Score"])

# 对数据进行分组
grouped_df = df.groupBy("Name")

# 对分组后的数据进行排序
sorted_df = grouped_df.sort(F.desc("Score"))

# 查看排序后的结果
sorted_df.show()

上述代码中，我们首先创建了一个 SparkSession 对象，并使用 createDataFrame 方法创建了一个 DataFrame 对象。然后，我们调用 groupBy 方法对数据进行分组，并使用 sort 方法对分组后的数据进行排序。最后，我们使用 show 方法查看排序后的结果。

执行以上代码后，我们将得到以下排序后的结果：

+-------+--------------+
|   Name|sort(desc(Score))|
+-------+--------------+
|    Bob|            95|
|  Alice|            90|
|  Carol|            75|
+-------+--------------+