DataFrame 删除另一个 DataFrame 中存在的行答案

【问题标题】：DataFrame remove rows existing in another DataFrameDataFrame 删除另一个 DataFrame 中存在的行
【发布时间】：2021-11-12 17:05:30
【问题描述】：

我有两个数据框：

df1:

+----------+-------------+-------------+--------------+---------------+
|customerId|     fullName|   telephone1|    telephone2|          email|
+----------+-------------+-------------+--------------+---------------+
|    201534|MARIO JIMENEZ|01722-3500391|+5215553623333|ascencio@my.com|
|    879535|  MARIO LOPEZ|01722-3500377|+5215553623333| asceloe@my.com|
+----------+-------------+-------------+--------------+---------------+

df2:

+----------+-------------+-------------+--------------+---------------+
|customerId|     fullName|   telephone1|    telephone2|          email|
+----------+-------------+-------------+--------------+---------------+
|    201534|MARIO JIMENEZ|01722-3500391|+5215553623333|ascencio@my.com|
|    201536|  ROBERT MITZ|01722-3500377|+5215553623333| asceloe@my.com|
|    201537|     MARY ENG|01722-3500127|+5215553623111|generic1@my.com|
|    201538|    RICK BURT|01722-3500983|+5215553623324|generic2@my.com|
|    201539|     JHON DOE|01722-3502547|+5215553621476|generic3@my.com|
+----------+-------------+-------------+--------------+---------------+

我需要从 df1 中获取第三个 DataFrame，这些 DataFrame 在 df2 中不存在。

像这样：

+----------+-------------+-------------+--------------+---------------+
|customerId|     fullName|   telephone1|    telephone2|          email|
+----------+-------------+-------------+--------------+---------------+
|    879535|  MARIO LOPEZ|01722-3500377|+5215553623333| asceloe@my.com|
+----------+-------------+-------------+--------------+---------------+

这样做的正确方法是什么？

我已经尝试过以下方法：

diff = df2.join(df1, df2['customerId'] != df1['customerId'],"left")

diff = df1.subtract(df2)

diff = df1[~ df1['customerId'].isin(df2['customerId'])]

但它们不起作用，有什么建议吗？

【问题讨论】：

一般来说，如果你能提供代码来生成你的数据框，人们会更容易提供帮助。
您的“喜欢这个”示例是 df2 中确实存在的示例，但是您说您的“需要”是“df2 中不存在的”请解决矛盾，否则我们不能这样。

标签： pandas dataframe pyspark aws-glue aws-glue-spark

【解决方案1】：

您可以将merge 与indicator=True 一起使用：

df3 = df1.merge(df2, on=df1.columns.tolist(), how='left', indicator=True)
df3 = df3[df3['_merge'] == 'left_only'].drop(columns='_merge')

输出：

>>> df3
   customerId     fullName     telephone1     telephone2           email
1      879535  MARIO LOPEZ  01722-3500377  5215553623333  asceloe@my.com

【讨论】：

【解决方案2】：

使用pyspark：

您可以创建一个包含来自DF2 和collect() 的customerId 的列表：

from pyspark.sql.types import *
id_df2 = [id[0] for id in df2.select('customerId').distinct().collect()]

然后过滤您的DF1 customerId 使用isin 否定~：

diff = df1.where(~col('customerId').isin(id_df2))

【讨论】：

我只需要导入 from pyspark.sql import functions as F 并将其用作 F.col() 但这很有效，谢谢
您的解决方案是杀死 Spark 的驱动程序节点的秘诀。有一个简单的解决方案：df1.join(df2, on='customerId', how='left_anti')
感谢您的反馈@Kafels。我经常使用上面的命令，所以你能详细说明一下为什么它不是一个好习惯吗？
@sophocles collect() 会将所有数据从 Workers 移动到 Driver 节点，在您有大量 GB 数据的示例中，这可能会导致 OOM 异常，这意味着您的整个集群将会消失下。通常，您确实需要从 spark 收集或移动一些数据到 python 上下文以应用特定逻辑的用例应该很少见。否则，如果您正在使用绘图数据，则无法避免