Pyspark按另一个数据帧的列过滤数据帧

Question

我不知道为什么我会遇到困难,看起来很简单,因为在R或熊猫中相当容易.我想避免使用pandas,因为我正在处理大量数据,我相信toPandas()所有数据都会加载到pyspark中的驱动程序内存中.

我有2个数据帧:df1和df2.我想过滤df1(删除所有行)df1.userid = df2.useridAND df1.group = df2.group.我不知道我是否应该使用filter(),join()或sql 例如:

df1:
+------+----------+--------------------+
|userid|   group  |      all_picks     |
+------+----------+--------------------+
|   348|         2|[225, 2235, 2225]   |
|   567|         1|[1110, 1150]        |
|   595|         1|[1150, 1150, 1150]  |
|   580|         2|[2240, 2225]        |
|   448|         1|[1130]              |
+------+----------+--------------------+

df2:
+------+----------+---------+
|userid|   group  |   pick  |
+------+----------+---------+
|   348|         2|     2270|
|   595|         1|     2125|
+------+----------+---------+

Result I want:
+------+----------+--------------------+
|userid|   group  |      all_picks     |
+------+----------+--------------------+
|   567|         1|[1110, 1150]        |
|   580|         2|[2240, 2225]        |
|   448|         1|[1130]              |
+------+----------+--------------------+

编辑:我已经尝试了很多join()和filter()函数,我相信我得到的最接近的是:

cond = [df1.userid == df2.userid, df2.group == df2.group]
df1.join(df2, cond, 'left_outer').select(df1.userid, df1.group, df1.all_picks) # Result has 7 rows

我尝试了一堆不同的连接类型,我也尝试了不同的cond值:cond =((df1.userid == df2.userid)&(df2.group == df2.group))#result有7行cond =(( df1.userid!= df2.userid)&(df2.group!= df2.group))#result有2行

但是,似乎联接正在添加其他行,而不是删除.

我正在使用python 2.7和spark 2.1.0

Answer 1

左反连接是你正在寻找的:

df1.join(df2, ["userid", "group"], "leftanti")

但左外连接可以做同样的事情:

(df1
    .join(df2, ["userid", "group"], "leftouter")
    .where(df2["pick"].isNull())
    .drop(df2["pick"]))