如何对比PySpark中含多值行的DataFrame与单值DataFrame？

核心思路 #

先提取df2的有效匹配值集合，对df1的每行字符串按点拆分后，优先匹配canada，若未匹配到则从左到右遍历拆分后的元素，返回第一个在有效集合中的值，无匹配则返回NULL。

完整代码实现 #

1. 初始化数据（示例）

from pyspark.sql import SparkSession
from pyspark.sql.functions import split, col, udf, broadcast
from pyspark.sql.types import StringType

spark = SparkSession.builder.appName("ValueMatch").getOrCreate()

# 构建示例df1
df1_data = [
    ("canada",),
    ("canada",),
    ("canada.Italy.Rome",),
    ("canada.USA.Brazil",),
    ("USA.Brazil",),
    ("canada.Rome",),
    ("France",),
    ("Japan.China",)
]
df1 = spark.createDataFrame(df1_data, ["xyz"])

# 构建示例df2
df2_data = [("canada",), ("USA",), ("Italy",), ("Rome",)]
df2 = spark.createDataFrame(df2_data, ["value"])

2. 准备匹配资源

将df2的有效值转为广播集合（小数据集广播可提升匹配效率）：

# 提取df2所有有效值并广播
valid_values = set(df2.select("value").rdd.flatMap(lambda x: x).collect())
broadcast_valid = spark.sparkContext.broadcast(valid_values)

3. 实现匹配逻辑UDF

def match_target(xyz_str):
    parts = xyz_str.split(".")
    for part in parts:
        # 优先匹配canada
        if part == "canada":
            return "canada"
        # 匹配其他有效值
        if part in broadcast_valid.value:
            return part
    # 无匹配返回NULL
    return None

# 注册UDF
match_udf = udf(match_target, StringType())

4. 生成结果

result_df = df1.withColumn("xyz_lst", match_udf(col("xyz")))
result_df.show(truncate=False)

输出结果 #

+-------------------+-------+
|xyz                |xyz_lst|
+-------------------+-------+
|canada             |canada |
|canada             |canada |
|canada.Italy.Rome  |canada |
|canada.USA.Brazil  |canada |
|USA.Brazil         |USA    |
|canada.Rome        |canada |
|France             |null   |
|Japan.China        |null   |
+-------------------+-------+

对现有代码的优化说明 #

1. 无需自定义拆分UDF：PySpark内置split(col("xyz"), "\\.")即可将字符串拆分为数组，替代你写的tryu函数。
2. 无需按数组长度拆分DataFrame：通过UDF或内置高阶函数可统一处理所有长度的拆分结果，无需分表处理。
3. 广播小数据集：将df2的有效值广播到每个executor，避免重复查询，提升匹配性能。

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