PySpark 如何在某些匹配条件下进行 LEFT ANTI join

在本文中，我们将介绍如何使用PySpark在某些匹配条件下进行LEFT ANTI join操作。

LEFT ANTI join简介

在PySpark中，LEFT ANTI join是关系型数据库中的一种连接操作。它返回仅在左侧数据集中出现而不在右侧数据集中出现的记录。这意味着我们可以使用LEFT ANTI join来筛选掉左边数据集中满足某些匹配条件的记录。

PySpark实现LEFT ANTI join

首先，我们需要创建两个数据集来演示LEFT ANTI join的用法。假设我们有一个员工表(employee)和一个部门表(department)。我们希望使用员工表中的一列作为匹配条件，筛选出不在部门表中的员工记录。

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("LEFT ANTI join").getOrCreate()

employee_data = [("John", "IT"), ("Mike", "HR"), ("Alice", "IT"), ("Tom", "Finance")]
employee_columns = ["name", "department"]
employee_df = spark.createDataFrame(data=employee_data, schema=employee_columns)

department_data = [("IT", "Information Technology"), ("HR", "Human Resources")]
department_columns = ["department", "department_name"]
department_df = spark.createDataFrame(data=department_data, schema=department_columns)

现在我们有了两个数据集，employee_df和department_df。接下来，我们将通过LEFT ANTI join操作筛选出不在部门表中的员工记录。

from pyspark.sql.functions import col

left_anti_join_df = employee_df.join(department_df, on=["department"], how="left_anti")

left_anti_join_df.show()

执行以上代码，我们将得到不在部门表中的员工记录：

+-----+----------+
| name|department|
+-----+----------+
| John|        IT|
| Alice|        IT|
|  Tom|   Finance|
+-----+----------+

在这个示例中，由于”Mike”作为一个HR部门的员工记录在部门表中有匹配项，因此在LEFT ANTI join操作中被筛选掉了。

除了简单地根据相等条件进行LEFT ANTI join，我们还可以在条件表达式中使用其他比较运算符，以满足特定的匹配需求。让我们考虑一个更复杂的例子，假设我们要筛选出员工表中工资高于部门表中平均工资的员工记录。

首先，我们需要添加一个额外的列salary到两个数据集中。

employee_data = [("John", "IT", 5000), ("Mike", "HR", 4000), ("Alice", "IT", 6000), ("Tom", "Finance", 7000)]
employee_columns = ["name", "department", "salary"]
employee_df = spark.createDataFrame(data=employee_data, schema=employee_columns)

department_data = [("IT", "Information Technology", 5500), ("HR", "Human Resources", 4000)]
department_columns = ["department", "department_name", "average_salary"]
department_df = spark.createDataFrame(data=department_data, schema=department_columns)

现在我们有了更新后的员工表employee_df和部门表department_df。接下来，我们将使用LEFT ANTI join操作筛选出工资高于部门平均工资的员工记录。

left_anti_join_df = employee_df.join(department_df, (employee_df["department"] == department_df["department"]) & (employee_df["salary"] > department_df["average_salary"]), how="left_anti")

left_anti_join_df.show()

执行以上代码，我们将得到工资高于部门平均工资的员工记录：

+-----+----------+------+
| name|department|salary|
+-----+----------+------+
| John|        IT|  5000|
| Alice|        IT|  6000|
|  Tom|   Finance|  7000|
+-----+----------+------+

在这个示例中，我们使用了条件表达式(employee_df["department"] == department_df["department"]) & (employee_df["salary"] > department_df["average_salary"])来进行LEFT ANTI join操作。只有满足这些条件的员工记录才会被保留。