PySpark:使用 isin returns 空数据框进行过滤

PySpark: filtering with isin returns empty dataframe

上下文: 我需要使用 isin 函数根据包含另一个数据框列的内容过滤数据框。

对于使用 pandas 的 Python 用户,这将是:isin().
对于 R 用户,这将是:%in%.

所以我有一个带有 idvalue 列的简单 spark 数据框:

l = [(1, 12), (1, 44), (1, 3), (2, 54), (3, 18), (3, 11), (4, 13), (5, 78)]
df = spark.createDataFrame(l, ['id', 'value'])
df.show()

+---+-----+
| id|value|
+---+-----+
|  1|   12|
|  1|   44|
|  1|    3|
|  2|   54|
|  3|   18|
|  3|   11|
|  4|   13|
|  5|   78|
+---+-----+

我想获取所有出现多次的id。这是 df 中唯一 ID 的数据框:

unique_ids = df.groupBy('id').count().where(col('count') < 2)
unique_ids.show()

+---+-----+
| id|count|
+---+-----+
|  5|    1|
|  2|    1|
|  4|    1|
+---+-----+

所以逻辑操作是:

 df = df[~df.id.isin(unique_ids.id)]
 # This is the same than:
 df = df[df.id.isin(unique_ids.id) == False]

但是,我得到一个空数据框:

df.show()

+---+-----+
| id|value|
+---+-----+
+---+-----+ 

这个"error"以相反的方式工作:

df[df.id.isin(unique_ids.id)]

returns df 的所有行。

表达式 df.id.isin(unique_ids.id) == False 正在计算是否 Column<b'((id IN (id)) = false)'>,这永远不会发生,因为 id 在 id 中。但是,表达式 df.id.isin(unique_ids.id) 正在评估 if Column<b'(id IN (id))'>,这始终为真,因此它 returns 整个数据框。 unique_ids.id 是列而不是列表。

isin(*cols) 接收一个 值列表 作为参数,而不是列,因此,要以这种方式工作,您应该执行以下操作:

ids = unique_ids.rdd.map(lambda x:x.id).collect()
df[df.id.isin(ids)].collect() # or show...

你将获得:

[Row(id=2, value=54), Row(id=4, value=13), Row(id=5, value=78)]

无论如何,我认为如果你加入两个数据框会更好:

df_ = df.join(unique_ids, on='id')

得到:

df_.show()
+---+-----+-----+
| id|value|count|
+---+-----+-----+
|  5|   78|    1|
|  2|   54|    1|
|  4|   13|    1|
+---+-----+-----+