如何高效比较写法不一致的两列国家名称（Pandas/Spark方案均可）

方法1：标准化映射表（准确率优先） #

先把各种不规范的国家名称统一映射到权威标准（比如ISO 3166官方名称），再做合并对比，适合长期维护：

• 先构建映射字典（可以存成CSV批量导入，方便维护上百个国家）：

  country_mapping = {
      "USA": "United States",
      "U.S.A.": "United States",
      "UK": "United Kingdom",
      "Britain": "United Kingdom",
      "China": "People's Republic of China",
      # 补充所有你遇到的变体
  }
  # 批量导入写法：
  # country_mapping = pd.read_csv("country_mappings.csv").set_index("variant")["standard"].to_dict()
  

• 对两个DataFrame的country列做标准化：

  df_1["standard_country"] = df_1["country"].map(country_mapping).fillna(df_1["country"])
  df_2["standard_country"] = df_2["country"].map(country_mapping).fillna(df_2["country"])
  

• 合并后对比标准化列：

  merged_df = pd.merge(df_1, df_2, on=["col_1", "col_2"], suffixes=("_1", "_2"))
  merged_df["country_match"] = merged_df["standard_country_1"] == merged_df["standard_country_2"]
  

• 针对未覆盖的变体，用模糊匹配补全：
  先装依赖pip install fuzzywuzzy python-Levenshtein，再写函数匹配相似名称：

  from fuzzywuzzy import process
  
  standard_countries = list(country_mapping.values())
  def fuzzy_standardize(country):
      if country in standard_countries:
          return country
      match, score = process.extractOne(country, standard_countries)
      return match if score >= 80 else country  # 相似度阈值可调整
  
  df_1["standard_country"] = df_1["country"].apply(fuzzy_standardize)
  df_2["standard_country"] = df_2["country"].apply(fuzzy_standardize)
  

方法2：直接模糊匹配合并（快速验证） #

不想维护映射表的话，直接在分组内做模糊匹配，适合临时需求：

from fuzzywuzzy import fuzz

# 按col_1、col_2分组，组内匹配相似国家名
def match_group(group):
    df1_rows = group[group["source"] == "df1"]
    df2_rows = group[group["source"] == "df2"]
    matches = []
    for _, r1 in df1_rows.iterrows():
        for _, r2 in df2_rows.iterrows():
            if fuzz.ratio(r1["country"], r2["country"]) >= 80:
                matches.append({**r1, **r2, "match_score": fuzz.ratio(r1["country"], r2["country"])})
    return pd.DataFrame(matches)

# 标记来源后分组匹配
df_1["source"] = "df1"
df_2["source"] = "df2"
combined = pd.concat([df_1, df_2])
result = combined.groupby(["col_1", "col_2"]).apply(match_group).reset_index(drop=True)
# 组内没匹配到的就是真正不匹配的记录

方法1：广播映射表做标准化 #

把映射表广播到集群节点，避免重复传输：

import org.apache.spark.sql.functions._

// 构建映射字典
val countryMapping = Map(
  "USA" -> "United States",
  "UK" -> "United Kingdom",
  "Britain" -> "United Kingdom"
)
val broadcastMapping = spark.sparkContext.broadcast(countryMapping)

// 定义标准化UDF
val standardizeCountry = udf((country: String) => {
  broadcastMapping.value.getOrElse(country, country)
})

// 标准化后合并对比
val df1Standard = df_1.withColumn("standard_country", standardizeCountry(col("country")))
val df2Standard = df_2.withColumn("standard_country", standardizeCountry(col("country")))

val mergedDf = df1Standard.join(df2Standard, Seq("col_1", "col_2"), "inner")
  .withColumn("country_match", col("standard_country") === col("standard_country"))

方法2：内置编辑距离匹配 #

用Spark自带的levenshtein函数计算字符差异，设置阈值判断匹配：

// 合并后计算编辑距离，阈值根据需求调整
val mergedDf = df_1.join(df_2, Seq("col_1", "col_2"), "inner")
  .withColumn("edit_distance", levenshtein(col("country"), col("country")))
  .withColumn("country_match", col("edit_distance") <= 3)

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