PySpark多列重复/相似度检测：求占比及跨表对比矩阵方案

一、核心流程 #

1. 加载所有客户数据表，给列添加表名前缀避免关联冲突
2. 生成所有表的两两组合，对每对表做关联计算
3. 针对字符串/编码类列选择对应相似度算法
4. 计算单列相似度后，按业务需求加权得到整体相似度
5. 聚合结果生成表级对比矩阵

二、列级相似度计算实现 #

1. 字符串列（name、street、city、country） #

基于编辑距离（Levenshtein）计算相似度，先统一清洗字符串（去空格、转小写），再转换为百分比：

from pyspark.sql import functions as F

def string_similarity(col1: str, col2: str) -> F.Column:
    # 清洗字符串：去首尾空格、转小写
    cleaned_col1 = F.trim(F.lower(F.col(col1)))
    cleaned_col2 = F.trim(F.lower(F.col(col2)))
    # 计算编辑距离
    lev_dist = F.levenshtein(cleaned_col1, cleaned_col2)
    # 取最大长度避免除以0
    max_len = F.greatest(F.length(cleaned_col1), F.length(cleaned_col2), F.lit(1))
    # 转换为相似度百分比并保留两位小数
    similarity = (1 - lev_dist / max_len) * 100
    return F.round(similarity, 2).alias(f"{col1}_vs_{col2}_similarity")

2. 编码列（postal code） #

先转字符串处理，优先精确匹配，也可根据业务需求增加前缀匹配逻辑：

def postal_code_similarity(col1: str, col2: str) -> F.Column:
    cleaned_col1 = F.trim(F.col(col1).cast("string"))
    cleaned_col2 = F.trim(F.col(col2).cast("string"))
    # 精确匹配返回100%，否则0%
    exact_match = F.when(cleaned_col1 == cleaned_col2, 100.0).otherwise(0.0)
    # 可选：前缀匹配（比如前3位相同返回70%）
    # prefix_match = F.when(F.substring(cleaned_col1,1,3) == F.substring(cleaned_col2,1,3),70.0).otherwise(0.0)
    # 可组合逻辑：精确匹配优先，否则用前缀匹配
    # final_similarity = F.when(exact_match == 100.0, 100.0).otherwise(prefix_match)
    return exact_match.alias(f"{col1}_vs_{col2}_similarity")

三、多表两两对比与矩阵生成 #

1. 加载并预处理数据表 #

给每个表的列加上表名前缀，避免关联时列名冲突：

# 示例：替换为你的实际表名列表
table_names = ["plant_cust_2022", "plant_cust_2023", "plant_cust_supplier"]
dfs = {}
for name in table_names:
    # 加载表并给每个列添加表名前缀
    df = spark.table(name)
    renamed_df = df.withColumnRenamed("name", f"{name}_name")\
                   .withColumnRenamed("street", f"{name}_street")\
                   .withColumnRenamed("city", f"{name}_city")\
                   .withColumnRenamed("postal code", f"{name}_postal_code")\
                   .withColumnRenamed("country", f"{name}_country")
    dfs[name] = renamed_df

2. 两两表关联计算相似度 #

遍历所有表对，做笛卡尔积关联（大表场景需优化），计算每列及整体相似度：

from itertools import combinations

comparison_outputs = []
# 生成所有不重复的表对
for table_a, table_b in combinations(table_names, 2):
    df_a = dfs[table_a]
    df_b = dfs[table_b]
    
    # 关联两张表：大表场景建议用广播小表优化，或按country分区关联
    joined_df = df_a.crossJoin(F.broadcast(df_b))
    
    # 计算各列相似度+整体加权相似度（权重根据业务调整）
    similarity_df = joined_df.select(
        F.lit(table_a).alias("source_table"),
        F.lit(table_b).alias("target_table"),
        string_similarity(f"{table_a}_name", f"{table_b}_name"),
        string_similarity(f"{table_a}_street", f"{table_b}_street"),
        string_similarity(f"{table_a}_city", f"{table_b}_city"),
        postal_code_similarity(f"{table_a}_postal_code", f"{table_b}_postal_code"),
        string_similarity(f"{table_a}_country", f"{table_b}_country"),
        # 整体相似度：按业务权重加权平均
        F.round(
            (string_similarity(f"{table_a}_name", f"{table_b}_name") * 0.3 +
             string_similarity(f"{table_a}_street", f"{table_b}_street") * 0.25 +
             string_similarity(f"{table_a}_city", f"{table_b}_city") * 0.2 +
             postal_code_similarity(f"{table_a}_postal_code", f"{table_b}_postal_code") * 0.15 +
             string_similarity(f"{table_a}_country", f"{table_b}_country") * 0.1),
            2
        ).alias("overall_similarity")
    )
    comparison_outputs.append(similarity_df)

# 合并所有对比结果
final_result = spark.createDataFrame([], comparison_outputs[0].schema)
for df in comparison_outputs:
    final_result = final_result.union(df)

3. 生成表级对比矩阵 #

按表对分组计算平均相似度，可转成宽表格式的矩阵：

# 按表对分组，计算各列及整体的平均相似度
matrix_summary = final_result.groupBy("source_table", "target_table").agg(
    F.avg(f"{table_a}_name_vs_{table_b}_name_similarity").alias("avg_name_similarity"),
    F.avg(f"{table_a}_street_vs_{table_b}_street_similarity").alias("avg_street_similarity"),
    F.avg(f"{table_a}_city_vs_{table_b}_city_similarity").alias("avg_city_similarity"),
    F.avg(f"{table_a}_postal_code_vs_{table_b}_postal_code_similarity").alias("avg_postal_code_similarity"),
    F.avg(f"{table_a}_country_vs_{table_b}_country_similarity").alias("avg_country_similarity"),
    F.avg("overall_similarity").alias("avg_overall_similarity")
)

# 可选：转成宽表矩阵（行=source_table，列=target_table）
wide_matrix = matrix_summary.groupBy("source_table").pivot("target_table").agg(
    F.first("avg_overall_similarity").alias("overall_similarity")
)

四、性能优化建议 #

• 广播小表：当其中一张表数据量远小于另一张时，用F.broadcast(df_b)减少Shuffle数据量
• 分区关联：如果表按country分区，可先按分区过滤再关联，避免全量笛卡尔积
• LSH近似匹配：超大规模表场景，用PySpark的MinHashLSH先筛选出高相似度候选记录，再做精确计算
• 采样计算：若不需要全量精确结果，可对每张表采样，用样本相似度近似整体情况

五、注意事项 #

• 空值处理：计算前用F.coalesce(cleaned_col, F.lit(""))替换空值，避免报错
• 业务权重调整：整体相似度的加权系数需根据业务优先级调整（比如name权重高于street）
• 邮编标准化：部分国家邮编含特殊字符/空格，需先做清洗（比如F.regexp_replace(cleaned_col, "\\s", "")）

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