PySpark：使用 isin 过滤返回空数据框

Posted 2023-04-15

【中文标题】PySpark：使用 isin 过滤返回空数据框

【英文标题】：PySpark: filtering with isin returns empty dataframe

【发布时间】：2019-03-07 14:41:15

【问题描述】：

背景： 我需要使用 isin 函数根据包含另一个数据框列的内容过滤数据框。

对于使用 pandas 的 Python 用户，这将是：isin()。 对于 R 用户，这将是：%in%。

所以我有一个带有 id 和 value 列的简单 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|
+---+-----+
+---+-----+ 

这个“错误”以相反的方式起作用：

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

返回df的所有行。

【问题讨论】：

不要在这里使用isin - 使用join。例如：df.join(unique_ids, on="id").show()。 isin 只能用于文字值（例如：df.where(df["id"].isin([1, 2, 3]))），不能用于列。

相关/可能的骗子：Pyspark isin function、PySpark: match the values of a DataFrame column against another DataFrame column、pyspark: isin vs join

【参考方案1】：

表达式df.id.isin(unique_ids.id) == False 正在评估Column<b'((id IN (id)) = false)'>，这永远不会发生，因为id 在id 中。但是，表达式df.id.isin(unique_ids.id) 正在评估Column<b'(id IN (id))'>，这始终是正确的，因此它返回整个数据帧。 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|
+---+-----+-----+

【讨论】：

“我觉得会更好” -> 会更好。

在这种情况下，为了我想要的，我会将 df 加入非唯一 ID 列表。

好吧，我错了，所以在 join how 参数中你必须指定一个how='left_anti' join:df_ = df.join(unique_ids, on='id', how='left_anti')，默认为how='inner'

谢谢，这也是我遇到的问题。就像你说的那样，我期待当你输入一列时，它会（逻辑上）输入列值。PySpark 文档在这方面可能会更加清晰......