含缺失值与格式差异的表关联方法咨询（SQL/Spark/Polars）

针对单字段缺失的关联需求，核心是分层匹配+兜底补全，以下是适配SQL、Spark、Polars的具体方案：

1. 多条件分层关联（按字段可靠性优先级匹配） #

优先用最具唯一性的字段（如company_name）关联，匹配失败的行再依次用次优先级字段（address→zipcode）尝试，最后合并所有匹配结果。这种方式能最大化利用有效字段，避免单一coalesce条件的误匹配。

Polars 实现示例 #

# 合并两个原始表
combined_df = df1.vstack(df2)

# 第一步：按company_name匹配（排除名称缺失的行）
match_by_name = combined_df.join(
    combined_df,
    on="company_name",
    how="left",
    suffix="_other"
).filter(pl.col("company_name").is_not_null())

# 第二步：处理名称缺失的行，按address匹配
unmatched_after_name = combined_df.filter(pl.col("company_name").is_null())
match_by_address = unmatched_after_name.join(
    combined_df,
    on="address",
    how="left",
    suffix="_other"
).filter(pl.col("address").is_not_null())

# 第三步：处理名称/地址都缺失的行，按zipcode匹配
unmatched_after_address = unmatched_after_name.filter(pl.col("address").is_null())
match_by_zip = unmatched_after_address.join(
    combined_df,
    on="zipcode",
    how="left",
    suffix="_other"
)

# 合并所有匹配结果并补全缺失字段
final_df = match_by_name.vstack(match_by_address).vstack(match_by_zip)
final_df = final_df.with_columns(
    pl.coalesce("company_name", "company_name_other").alias("cleaned_company"),
    pl.coalesce("address", "address_other").alias("cleaned_address"),
    pl.coalesce("zipcode", "zipcode_other").alias("cleaned_zipcode")
)

SQL 实现示例 #

-- 按company_name匹配的结果
SELECT 
    COALESCE(t1.company_name, t2.company_name) AS cleaned_company,
    COALESCE(t1.address, t2.address) AS cleaned_address,
    COALESCE(t1.zipcode, t2.zipcode) AS cleaned_zipcode
FROM combined_table t1
LEFT JOIN combined_table t2 ON t1.company_name = t2.company_name
WHERE t1.company_name IS NOT NULL

UNION ALL

-- 名称缺失时，按address匹配的结果
SELECT 
    COALESCE(t1.company_name, t2.company_name) AS cleaned_company,
    COALESCE(t1.address, t2.address) AS cleaned_address,
    COALESCE(t1.zipcode, t2.zipcode) AS cleaned_zipcode
FROM combined_table t1
LEFT JOIN combined_table t2 ON t1.address = t2.address
WHERE t1.company_name IS NULL AND t1.address IS NOT NULL

UNION ALL

-- 名称/地址都缺失时，按zipcode匹配的结果
SELECT 
    COALESCE(t1.company_name, t2.company_name) AS cleaned_company,
    COALESCE(t1.address, t2.address) AS cleaned_address,
    COALESCE(t1.zipcode, t2.zipcode) AS cleaned_zipcode
FROM combined_table t1
LEFT JOIN combined_table t2 ON t1.zipcode = t2.zipcode
WHERE t1.company_name IS NULL AND t1.address IS NULL

Spark 实现示例 #

// 合并两个原始表
val combinedDf = df1.union(df2)

// 按company_name匹配
val matchByName = combinedDf.join(combinedDf, Seq("company_name"), "left_outer")
  .filter(col("company_name").isNotNull)
  .select(
    coalesce(col("company_name"), col("company_name")).alias("cleaned_company"),
    coalesce(col("address"), col("address")).alias("cleaned_address"),
    coalesce(col("zipcode"), col("zipcode")).alias("cleaned_zipcode")
  )

// 处理名称缺失的行，按address匹配
val unmatchedAfterName = combinedDf.filter(col("company_name").isNull)
val matchByAddress = unmatchedAfterName.join(combinedDf, Seq("address"), "left_outer")
  .filter(col("address").isNotNull)
  .select(
    coalesce(col("company_name"), col("company_name")).alias("cleaned_company"),
    coalesce(col("address"), col("address")).alias("cleaned_address"),
    coalesce(col("zipcode"), col("zipcode")).alias("cleaned_zipcode")
  )

// 处理名称/地址都缺失的行，按zipcode匹配
val unmatchedAfterAddress = unmatchedAfterName.filter(col("address").isNull)
val matchByZip = unmatchedAfterAddress.join(combinedDf, Seq("zipcode"), "left_outer")
  .select(
    coalesce(col("company_name"), col("company_name")).alias("cleaned_company"),
    coalesce(col("address"), col("address")).alias("cleaned_address"),
    coalesce(col("zipcode"), col("zipcode")).alias("cleaned_zipcode")
  )

// 合并所有结果
val finalDf = matchByName.union(matchByAddress).union(matchByZip)

2. 分组填充补全 #

如果某行多个字段缺失，但同组（如同zipcode、同城市）的其他行有该公司的完整信息，可以通过分组窗口函数，用组内非空值填充缺失字段：

Polars 示例 #

final_df = combined_df.with_columns(
    pl.col("company_name").fill_null(pl.col("company_name").over("zipcode")),
    pl.col("address").fill_null(pl.col("address").over("zipcode"))
)

SQL 示例 #

SELECT 
    COALESCE(company_name, MAX(company_name) OVER (PARTITION BY zipcode)) AS cleaned_company,
    COALESCE(address, MAX(address) OVER (PARTITION BY zipcode)) AS cleaned_address,
    zipcode
FROM combined_table

3. 模糊关联（兜底方案） #

当精确匹配失败时，可对非空字段做模糊匹配（如地址包含关键词、名称去掉后缀后匹配），降低缺失值导致的匹配失败率：

Polars 模糊匹配示例 #

unmatched_rows = combined_df.filter(pl.col("company_name").is_null())
fuzzy_match = unmatched_rows.join(
    combined_df,
    on=pl.col("address").str.contains(pl.col("address_other"), regex=False) | 
       pl.col("address_other").str.contains(pl.col("address"), regex=False),
    how="left",
    suffix="_other"
)

格式差异会导致有效字段无法匹配，建议先做标准化预处理，再进行关联：

• 公司名称：统一大小写、去掉后缀（Inc./Corp.）、清理特殊字符

  # Polars 名称标准化
  combined_df = combined_df.with_columns(
      pl.col("company_name").str.replace(r"\s*(Inc\.|Corp\.|Ltd\.)$", "", literal=False)
        .str.to_lowercase()
        .alias("standardized_name")
  )
  

• 地址：提取核心街道信息、去掉冗余后缀（Suite X/Floor X）、统一格式

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