PySpark PySpark – 比较 DataFrames

在本文中，我们将介绍如何使用 PySpark 比较 DataFrames。PySpark 是一个基于 Apache Spark 的 Python 库，可以分布式处理大规模数据集。比较 DataFrames 可以帮助我们发现两个数据集之间的差异以及相似之处。

阅读更多：PySpark 教程

什么是 DataFrame?

在开始介绍如何比较 DataFrames 之前，我们先来了解一下什么是 DataFrame。DataFrame 是一个类似于关系型数据库中的表格的数据结构，它包含了一系列的行和列。每一列都有一个名称和一个数据类型，而每一行都有一个唯一的标识符。DataFrame 可以由各种数据源创建，如 CSV 文件、数据库表格以及 JSON 数据等。

在 PySpark 中，我们可以使用 spark.read 方法来读取数据源并创建 DataFrame，如下所示：

from pyspark.sql import SparkSession

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

# 从 CSV 文件创建 DataFrame
df1 = spark.read.csv("data1.csv", header=True, inferSchema=True)

# 从数据库表格创建 DataFrame
df2 = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/mydb").option("dbtable", "table1").option("user", "root").option("password", "password").load()

# 从 JSON 数据创建 DataFrame
df3 = spark.read.json("data3.json")

创建了 DataFrame 后，我们就可以使用各种操作来处理数据了，比如筛选、排序、聚合等。

比较 DataFrames

在实际工作中，我们经常需要比较两个数据集之间的不同之处。使用 PySpark 可以简化这个过程。下面是一些常用的比较操作：

1. 判断两个 DataFrame 是否相等

要判断两个 DataFrame 是否相等，可以使用 DataFrame.equals 方法。这个方法返回一个布尔值，表示两个 DataFrame 是否相等。例如：

df1.equals(df2)

2. 查找不同的行

要找出两个 DataFrame 中不同的行，可以使用 DataFrame.subtract 方法。这个方法返回一个新的 DataFrame，包含了存在于第一个 DataFrame 但不存在于第二个 DataFrame 中的行。例如：

diff_rows = df1.subtract(df2)

3. 查找差异的列

要找出两个 DataFrame 中差异的列，可以使用 DataFrame.subtractAll 方法。这个方法返回一个新的 DataFrame，包含了第一个 DataFrame 和第二个 DataFrame 之间的列差异。例如：

diff_cols = df1.subtractAll(df2)

4. 查找重复的行

要找出一个 DataFrame 中重复的行，可以使用 DataFrame.dropDuplicates 方法。这个方法返回一个新的 DataFrame，其中包含了去除了重复行之后的数据。例如：

distinct_rows = df.dropDuplicates()

5. 查找相似的行

要找出两个 DataFrame 中相似的行，可以使用 DataFrame.join 方法。这个方法将两个 DataFrame 进行连接，并返回一个新的 DataFrame，其中包含了两个 DataFrame 中相似的行。例如：

similar_rows = df1.join(df2, df1["column1"] == df2["column1"], "inner")

示例

假设我们有两个数据集，分别包含了学生的姓名和年龄。我们想要比较两个数据集之间的差异，并找到相同的学生。下面是一个示例代码：

from pyspark.sql import SparkSession

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

# 从 CSV 文件创建 DataFrame
df1 = spark.read.csv("students1.csv", header=True, inferSchema=True)
df2 = spark.read.csv("students2.csv", header=True, inferSchema=True)

# 找出不同的行
diff_rows = df1.subtract(df2)

# 找出相同的行
similar_rows = df1.join(df2, df1["name"] == df2["name"], "inner")

# 输出结果
print("不同的行：")
diff_rows.show()

print("相同的行：")
similar_rows.show()

使用上述代码，我们可以找出两个数据集之间的差异，并找到相同的学生。