PySpark 在DataFrame中删除重复行

PySpark 是由 Apache Spark 社区设计的工具，用于实时处理数据并在本地Python环境中分析结果。Spark数据帧与其他数据帧不同，它将信息分布并遵循模式。

Spark既可以处理流处理，也可以处理批处理，这就是它们受欢迎的原因。PySpark数据帧需要会话以生成入口点，并在系统中对数据进行处理（RAM）。您可以使用以下命令在Windows上安装PySpark模块：

pip install pyspark

在本文中，我们将创建一个PySpark数据框并讨论从该数据框中删除重复行的不同方法。让我们了解一下PySpark数据框的概念。

创建和了解PySpark数据框

就像任何其他数据框一样，PySpark以表格形式存储数据。它允许程序员处理结构化和半结构化数据，并提供高级API（python，Java）以处理复杂的数据集。它可以快速分析数据，因此在流处理和批处理中非常有帮助。

现在我们已经讨论了PySpark数据框的基础知识，让我们使用Python代码来创建一个。我们将创建一个包含有关不同赛车手的信息的PySpark数据框。

示例

• 我们导入了必要的库，包括“pandas和“pyspark”。我们还导入了一个名为“SparkSession”的统一接口。

• 这个接口确保Spark框架正常运行。它充当引入多个Spark API的“入口点”，从而提高数据处理的效率。简而言之，我们创建一个SparkSession来设置所需的配置。

• 在完成配置部分之后，我们使用“builder” API创建了一个SparkSession的实例。我们还使用了“getorCreate()”方法来锁定现有会话或替换为新会话。

• 完成配置部分之后，我们准备了一个包含不同汽车特征的数据集字典。我们使用这个数据集生成了一个pandas数据框。

• 生成的5X3数据框存储在“dataframe_pd”变量中。此变量作为“SparkSession”的“createDataFrame()”方法的参数传递，以创建一个PySpark数据框。

• 我们使用一个pandas数据框生成了一个PySpark数据框，但这不是一个强制性的步骤。我们可以直接使用一个元组列表来创建一个数据集，然后将其传递给“createDataFrame()”方法。

• 最后，我们使用“dataframe_spk.show()”方法显示了数据框。

示例

import pyspark
from pyspark.sql import SparkSession
import pandas as pd
spark = SparkSession.builder.getOrCreate()
dataset = {"Carname":["Audi", "Mercedes", "BMW", "Audi", "Audi"], "Max Speed": ["300 KPH", "250 KPH", "220 KPH", "300 KPH", "300 KPH"], "Car number":["MS321", "QR345", "WX281", "MS321", "MS321"]}
dataframe_pd = pd.DataFrame(dataset, index= ["Racer1", "Racer2", "Racer3", "Racer1", "Racer1"])
dataframe_spk = spark.createDataFrame(dataframe_pd)
print("The original data frame is like: -")
dataframe_spk.show()

输出

The original data frame is like: -
    Carname Max Speed Car number
      Audi   300 KPH      MS321
  Mercedes   250 KPH      QR345
       BMW   220 KPH      WX281
      Audi   300 KPH      MS321
      Audi   300 KPH      MS321

现在我们已经创建了一个PySpark数据框，让我们讨论从该数据框中删除行的不同方法。

使用Distinct()函数删除行

该函数返回一个具有不同或唯一行的新数据框。它消除数据框中的所有重复行。

示例

我们不为这个函数传递任何参数。让我们看看它的实现。

from pyspark.sql import SparkSession
import pandas as pd

spark = SparkSession.builder.getOrCreate()

dataset = {"Carname":["Audi", "Mercedes", "BMW", "Audi", "Audi"], "Max Speed": ["300 KPH", "250 KPH", "220 KPH", "300 KPH", "300 KPH"], "Car number":["MS321", "QR345", "WX281", "MS321", "MS321"]}
dataframe_pd = pd.DataFrame(dataset)
dataframe_spk = spark.createDataFrame(dataframe_pd)
print("The original data frame is like: -")
dataframe_spk.show()
print("After dropping the duplicate rows we get: -")
dataframe_spk.distinct().show()

输出

The original data frame is like: -
    Carname Max Speed Car number
      Audi   300 KPH      MS321
  Mercedes   250 KPH      QR345
       BMW   220 KPH      WX281
      Audi   300 KPH      MS321
      Audi   300 KPH      MS321
After dropping the duplicate rows we get: -

    Carname Max Speed Car number
  Mercedes  250 KPH      QR345
      BMW   220 KPH      WX281
      Audi  300 KPH      MS321

创建PySpark数据框后，我们使用 distinct() 函数来定位唯一行并从数据框中消除它们。

使用dropDuplicate()函数

这是一种可替代的方法，与 distinct() 函数的功能相同。我们可以根据列来定位并删除相应的行。让我们来构建代码。

示例

from pyspark.sql import SparkSession
import pandas as pd

spark = SparkSession.builder.getOrCreate()

dataset = {"Carname":["Audi", "Mercedes", "BMW", "Audi", "Audi"], "Max Speed": ["300 KPH", "250 KPH", "220 KPH", "300 KPH", "300 KPH"], "Car number":["MS321", "QR345", "WX281", "MS321", "MS321"]}
dataframe_pd = pd.DataFrame(dataset)
dataframe_spk = spark.createDataFrame(dataframe_pd)
print("The original data frame is like: -")
dataframe_spk.show()

print("After dropping the duplicate rows we get: -")
dataframe_spk.dropDuplicates().show()

输出

The original data frame is like: -
+--------+---------+----------+
| Carname|Max Speed|Car number|
+--------+---------+----------+
|    Audi|  300 KPH|     MS321|
|Mercedes|  250 KPH|     QR345|
|     BMW|  220 KPH|     WX281|
|    Audi|  300 KPH|     MS321|
|    Audi|  300 KPH|     MS321|
+--------+---------+----------+
After dropping the duplicate rows we get: -
+--------+---------+----------+
| Carname|Max Speed|Car number|
+--------+---------+----------+
|Mercedes|  250 KPH|     QR345|
|    Audi|  300 KPH|     MS321|
|     BMW|  220 KPH|     WX281|
+--------+---------+----------+

定位特定列

我们可以使用“ select() ”函数来检查特定列中的重复值。我们将对选择的列使用dropDuplicate()函数。

示例

dataframe_spk.select(["Carname"]).dropDuplicates().show()

输出

+--------+
| Carname|
+--------+
|Mercedes|
|     BMW|
|    Audi|
+--------+

结论

这篇文章解释了如何删除“column”值中包含任何重复数据的行的基本操作。我们讨论了涉及的不同函数，包括“dropDuplicate()”，“Distinct()”和“select()”。我们创建了一个参考数据框，并从中删除了重复值。