如何在基于另一个 DataFrame 的列上删除 DataFrame 中的行？答案

【问题标题】：How to remove rows in DataFrame on column based on another DataFrame?如何在基于另一个 DataFrame 的列上删除 DataFrame 中的行？
【发布时间】：2016-11-14 11:47:09
【问题描述】：

我正在尝试在 Spark 1.6.1 中使用 SQLContext.subtract() 根据另一个数据帧中的列从数据帧中删除行。举个例子：

from pyspark.sql import Row

df1 = sqlContext.createDataFrame([
    Row(name='Alice', age=2),
    Row(name='Bob', age=1),
]).alias('df1')

df2 = sqlContext.createDataFrame([
    Row(name='Bob'),
])

df1_with_df2 = df1.join(df2, 'name').select('df1.*')
df1_without_df2 = df1.subtract(df1_with_df2)

由于我想要 df1 中不包括 name='Bob' 的所有行，我希望 Row(age=2, name='Alice')。但我也找回了 Bob：

print(df1_without_df2.collect())
# [Row(age='1', name='Bob'), Row(age='2', name='Alice')]

经过各种实验以了解这个MCVE，我发现问题出在age 键上。如果我省略它：

df1_noage = sqlContext.createDataFrame([
    Row(name='Alice'),
    Row(name='Bob'),
]).alias('df1_noage')

df1_noage_with_df2 = df1_noage.join(df2, 'name').select('df1_noage.*')
df1_noage_without_df2 = df1_noage.subtract(df1_noage_with_df2)
print(df1_noage_without_df2.collect())
# [Row(name='Alice')]

那我只能按预期得到爱丽丝。我所做的最奇怪的观察是，可以添加键，只要它们在之后（在字典顺序意义上）我在连接中使用的键：

df1_zage = sqlContext.createDataFrame([
    Row(zage=2, name='Alice'),
    Row(zage=1, name='Bob'),
]).alias('df1_zage')

df1_zage_with_df2 = df1_zage.join(df2, 'name').select('df1_zage.*')
df1_zage_without_df2 = df1_zage.subtract(df1_zage_with_df2)
print(df1_zage_without_df2.collect())
# [Row(name='Alice', zage=2)]

我正确地得到了爱丽丝（和她的 zage）！在我的真实示例中，我对所有列都感兴趣，而不仅仅是name 之后的列。

【问题讨论】：

标签： apache-spark pyspark apache-spark-sql

【解决方案1】：

这里有一些错误（第一个问题看起来与 SPARK-6231 相同的问题有关）并且 JIRA 看起来是个好主意，但 SUBTRACT / EXCEPT 不是部分匹配的正确选择.

相反，从Spark 2.0 开始，您可以使用反连接：

df1.join(df1_with_df2, ["name"], "leftanti").show()

在 1.6 中，您可以使用标准外连接做几乎相同的事情：

import pyspark.sql.functions as F

ref = df1_with_df2.select("name").alias("ref")

(df1
    .join(ref, ref.name == df1.name, "leftouter")
    .filter(F.isnull("ref.name"))
    .drop(F.col("ref.name")))

【讨论】：