PySpark：如何使用另一个DataFrame的列过滤目标DataFrame？

你原来的代码无法生效，是因为contains方法传入另一列时，Spark会尝试逐行两两比较DataFrame1和DataFrame2的对应行，而不是检查DataFrame1的列值是否包含DataFrame2列中的任意一个值，这不符合你的过滤需求。以下是三种可行的解决方法：

方法1：广播小DataFrame + Left Semi Join（推荐） #

适合DataFrame2数据量小（比如你的10行）的场景，利用广播避免笛卡尔积，效率高：

from pyspark.sql.functions import broadcast

# 使用leftsemi join，只保留DataFrame1中能匹配的行，不会引入DataFrame2的列
contract_con = dataframe1.join(
    broadcast(dataframe2),
    dataframe1.name_of_column.contains(dataframe2.name_of_column),
    "leftsemi"
).distinct()  # 去重避免重复匹配的行
contract_con.show()

方法2：广播变量 + exists函数 #

通过exists函数检查DataFrame1的列值是否包含DataFrame2中的任意一个值：

from pyspark.sql.functions import broadcast, exists, col

# 广播DataFrame2的过滤列
broadcast_filter = broadcast(dataframe2.select("name_of_column").alias("filter_val"))

# 过滤出包含任意过滤值的行
contract_con = dataframe1.filter(
    exists(broadcast_filter, lambda x: col("name_of_column").contains(x.filter_val))
)
contract_con.show()

方法3：拼接正则表达式（适合极小规模数据） #

把DataFrame2的列值拼成正则串，用rlike匹配（注意转义正则特殊字符）：

from pyspark.sql.functions import col, regexp_replace

# 提取DataFrame2的过滤值并转义正则特殊字符
filter_values = [row.name_of_column for row in dataframe2.collect()]
escaped_values = [regexp_replace(val, r'([\.\*\+\?\|\(\)\[\]\{\}\\])', r'\\$1') for val in filter_values]
pattern = "|".join(escaped_values)

# 用rlike匹配任意子串
contract_con = dataframe1.filter(col("name_of_column").rlike(pattern))
contract_con.show()

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