基于数组交集最高匹配度映射DataFrame列的PySpark/Pandas实现

先构造示例数据：

import pandas as pd
import numpy as np

df1 = pd.DataFrame({
    'A': ['AA', 'BB', 'CC'],
    'B': [['a', 'b', 'c', 'd'], ['a', 'f', 'g', 'c'], ['a', 'b', 'l', 'm']]
})

df2 = pd.DataFrame({
    'C': ['XX', 'YY', 'UU'],
    'D': [['a', 'b', 'c', 'n'], ['a', 'm', 'r', 's'], ['e', 'h', 'l', 'j']]
})

核心思路：生成df1和df2的笛卡尔积，遍历所有可能的匹配对计算交集，再按df2的C分组筛选交集元素最多的匹配项，无交集则设A为空。

完整代码：

# 生成笛卡尔积，临时标识列用于关联
df1['tmp_id'] = 1
df2['tmp_id'] = 1
cross_df = pd.merge(df1, df2, on='tmp_id').drop('tmp_id', axis=1)

# 定义函数计算公共元素和交集长度
def calc_common(list1, list2):
    common = list(set(list1) & set(list2))
    return common, len(common)

# 批量计算公共元素和匹配分数
cross_df[['common_items', 'match_score']] = cross_df.apply(
    lambda x: calc_common(x['B'], x['D']),
    axis=1,
    result_type='expand'
)

# 按C分组筛选最优匹配
def pick_best(group):
    max_score = group['match_score'].max()
    if max_score == 0:
        return pd.Series({'A': np.nan, 'common_items': []})
    # 取分数最高的第一行（若有多个同分可自行调整逻辑）
    best_row = group[group['match_score'] == max_score].iloc[0]
    return pd.Series({'A': best_row['A'], 'common_items': best_row['common_items']})

# 合并回原df2
result_df = cross_df.groupby('C').apply(pick_best).reset_index()
result_df = pd.merge(df2, result_df, on='C')[['C', 'D', 'A', 'common_items']]

print(result_df)

输出结果：

C                D     A common_items
0  XX  [a, b, c, n]    AA  [a, b, c]
1  YY  [a, m, r, s]    CC  [a, m]
2  UU  [e, h, l, j]  NaN           []

针对大数据量场景，用PySpark处理更高效，核心思路与Pandas一致，结合UDF和窗口函数实现：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, udf, row_number, when
from pyspark.sql.types import ArrayType, StringType, IntegerType, StructType, StructField
from pyspark.sql.window import Window

# 初始化Spark会话
spark = SparkSession.builder.appName("ListMatch").getOrCreate()

# 构造示例数据
df1 = spark.createDataFrame([
    ("AA", ["a", "b", "c", "d"]),
    ("BB", ["a", "f", "g", "c"]),
    ("CC", ["a", "b", "l", "m"])
], ["A", "B"])

df2 = spark.createDataFrame([
    ("XX", ["a", "b", "c", "n"]),
    ("YY", ["a", "m", "r", "s"]),
    ("UU", ["e", "h", "l", "j"])
], ["C", "D"])

# 定义UDF：返回公共元素列表和交集长度
def get_common(list1, list2):
    set1, set2 = set(list1), set(list2)
    common = list(set1 & set2)
    return (common, len(common))

# 定义UDF返回类型
udf_return_type = StructType([
    StructField("common_items", ArrayType(StringType())),
    StructField("match_score", IntegerType())
])

common_udf = udf(get_common, udf_return_type)

# 生成笛卡尔积并计算匹配信息
cross_df = df1.crossJoin(df2)
cross_df = cross_df.withColumn("match_info", common_udf(col("B"), col("D"))) \
                   .withColumn("common_items", col("match_info.common_items")) \
                   .withColumn("match_score", col("match_info.match_score")) \
                   .drop("match_info", "B")

# 窗口函数按C分组，按匹配分数降序排名
window_spec = Window.partitionBy("C").orderBy(col("match_score").desc())
ranked_df = cross_df.withColumn("rank", row_number().over(window_spec))

# 筛选最优匹配，无交集时设A为null
best_match_df = ranked_df.filter(col("rank") == 1) \
                         .withColumn("A", when(col("match_score") == 0, None).otherwise(col("A"))) \
                         .drop("rank", "match_score")

# 左连接回原df2，确保所有行保留
result_df = df2.join(best_match_df, on="C", how="left")

# 展示结果
result_df.show(truncate=False)

输出结果：

+---+----------------+----+------------+
|C  |D               |A   |common_items|
+---+----------------+----+------------+
|XX |[a, b, c, n]    |AA  |[a, b, c]   |
|YY |[a, m, r, s]    |CC  |[a, m]      |
|UU |[e, h, l, j]    |null|[]          |
+---+----------------+----+------------+

注：若存在多个df1行与同一df2行交集长度相同，上述代码默认取第一个匹配项，可根据需求调整pick_best函数或窗口函数的排序逻辑。

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