无需Join，如何基于另一PySpark DataFrame过滤列？

问题原因 #

你用isin()无效的核心原因是：这个方法需要传入具体的数值集合（比如列表、元组），而不是DataFrame的列对象。直接传df2.ID2_1这类Column类型参数，Spark无法解析出实际要匹配的值，所以过滤逻辑不生效。

可行解决方案（无需开销大的Join操作） #

方案1：提取合并后的唯一值列表（小数据量场景）

先从df2中把ID2_1和ID2_2的所有值合并、去重，得到一个值列表，再传入isin()：

from pyspark.sql import functions as f

# 合并df2的两列数据并去重，提取为Python列表
valid_ids = (df2.select(f.col("ID2_1").alias("id"))
             .union(df2.select(f.col("ID2_2").alias("id")))
             .distinct()
             .rdd.flatMap(lambda x: x)
             .collect())

# 过滤df1，保留ID2在有效列表中的行
filtered_df = df1.filter(f.col("ID2").isin(valid_ids))

注意：如果df2的数据量极大，collect()会把所有值拉到Driver端，可能引发内存溢出，这种情况不适用。

方案2：使用广播变量（小表过滤大表场景）

如果df2数据量不大，把有效id集合广播到所有Executor节点，减少Driver端压力：

# 获取去重的有效id列表
valid_ids = (df2.select(f.col("ID2_1").alias("id"))
             .union(df2.select(f.col("ID2_2").alias("id")))
             .distinct()
             .rdd.flatMap(lambda x: x)
             .collect())

# 创建广播变量
broadcast_valid_ids = spark.sparkContext.broadcast(valid_ids)

# 用广播变量的值过滤df1
filtered_df = df1.filter(f.col("ID2").isin(broadcast_valid_ids.value))

方案3：子查询过滤（大数据量场景）

无需把数据拉到Driver端，直接用Spark的子查询语法实现过滤，性能更优：

# 构造子查询，获取所有有效的ID值
valid_ids_subquery = (df2.select(f.col("ID2_1").alias("id"))
                      .union(df2.select(f.col("ID2_2").alias("id")))
                      .distinct())

# 使用exists子查询过滤
filtered_df = df1.filter(f.exists(valid_ids_subquery, f.col("id") == f.col("ID2")))

也可以用更简洁的in子查询写法：

filtered_df = df1.filter(f.col("ID2").isin(f.expr("SELECT ID2_1 FROM df2 UNION SELECT ID2_2 FROM df2")))

内容的提问来源于stack exchange，提问作者Ofek Glick