PySpark：在PySpark中聚合并删除重复项

在本文中，我们将介绍如何在PySpark中进行聚合操作，并且在聚合过程中删除重复项。PySpark是一个强大的分布式计算框架，可以处理大规模数据集。它提供了各种功能和操作，包括聚合和去重。

阅读更多：PySpark 教程

聚合操作

聚合是指将多个数据项组合为一个单一项的过程。在PySpark中，我们使用聚合函数来执行这些操作。常用的聚合函数包括sum、count、avg等。让我们看一个示例来理解如何使用聚合函数。

假设我们有一个包含学生成绩的数据集，包括学生姓名、科目和分数。我们想要计算每个科目的平均分数，并按科目进行分组。下面是一个示例数据集：

我们可以使用PySpark来计算每个科目的平均分数。首先，我们需要创建一个SparkSession对象：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("AggregateExample").getOrCreate()

然后，我们可以将数据集加载到一个DataFrame中：

df = spark.createDataFrame([
    ("小明", "数学", 90),
    ("小明", "英语", 85),
    ("小红", "数学", 95),
    ("小红", "英语", 92),
    ("小李", "数学", 88),
    ("小李", "英语", 90)
], ["姓名", "科目", "分数"])

接下来，我们可以使用groupBy和agg函数来进行聚合操作。我们首先按科目分组，然后计算每个组的平均分数：

df_grouped = df.groupBy("科目").agg({"分数": "avg"})

最后，我们可以查看结果：

df_grouped.show()

输出将会是：

+----+--------+
|科目  |avg(分数)|
+----+--------+
|数学  |91.0    |
|英语  |89.0    |
+----+--------+

如上所示，我们成功计算出了每个科目的平均分数。

删除重复项

在处理数据时，经常会出现重复的记录。在某些情况下，我们可能需要删除这些重复项。在PySpark中，我们可以使用dropDuplicates函数来删除重复项。让我们看一个示例来演示如何使用这个函数。

假设我们有一个包含多个城市的DataFrame，每个城市有不同的人口数量。但是由于数据采集的原因，数据集中可能存在重复记录。我们想要删除所有重复的城市记录，只保留每个城市的一条记录。

下面是一个示例数据集：

我们可以使用PySpark来删除重复的城市记录。首先，我们需要创建一个SparkSession对象：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("DropDuplicatesExample").getOrCreate()

然后，我们可以将数据集加载到一个DataFrame中：

df = spark.createDataFrame([
    ("北京", 2170),
    ("上海", 2424),
    ("广州", 1350),
    ("深圳", 1311),
    ("北京", 2170),
    ("上海", 2424)
], ["城市", "人口"])

接下来，我们可以使用dropDuplicates函数来删除重复的城市记录：

df_deduplicated = df.dropDuplicates()

最后，我们可以查看结果：

df_deduplicated.show()

输出将会是：

+----+--------+
|城市  |人口     |
+----+--------+
|北京  |2170    |
|上海  |2424    |
|广州  |1350    |
|深圳  |1311    |
+----+--------+

如上所示，我们成功删除了重复的城市记录。

总结

本文介绍了如何使用PySpark进行聚合操作，并且在聚合过程中删除重复项。我们学习了如何使用聚合函数来计算每个科目的平均分数，并使用dropDuplicates函数删除重复的城市记录。PySpark提供了强大的功能和操作，使我们能够高效地处理大规模数据集。通过掌握这些技巧，我们可以更好地利用PySpark进行数据处理和分析。