PySpark使用country-converter将DataFrame国家名转为ISO3代码

原方案通过collect()拉取全量数据到Driver节点转换的方式存在内存溢出风险，同时会丢失原表其他字段，可通过Pandas UDF分布式执行的方式一次性解决列保留、转换效率、异常值处理三个问题。

核心实现（保留原表所有字段） #

Pandas UDF按分区分布式执行转换逻辑，无需将全量数据拉取到Driver，可直接在原DataFrame基础上替换目标列，完整保留UserName等非目标字段：

import country_converter as coco
import pandas as pd
import re
from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import StringType

# 全局只初始化一次转换器，避免重复加载规则资源损耗性能
cc = coco.CountryConverter()
OUT_FORMAT = "ISO3"

@pandas_udf(StringType())
def loc_to_iso3(loc_series: pd.Series) -> pd.Series:
    convert_result = []
    for raw_loc in loc_series:
        # 统一转字符串，处理非字符串类型的异常输入
        loc_str = str(raw_loc).strip()
        convert_result.append(
            cc.convert(names=loc_str, to=OUT_FORMAT, not_found="UNMATCHED", enforce_list=False)
        )
    return pd.Series(convert_result)

# 直接替换原Location列，所有原表字段完整保留
result_df = df.withColumn("Location", loc_to_iso3("Location"))

执行后样例数据输出如下：

异常值批量处理方案 #

你遇到的Row(Countries='London, UK')类异常，本质是之前RDD转List时未提取Row内的实际字符串值，叠加部分地址是「城市,国家」的非标准格式导致的，不需要逐值手动replace，在UDF内增加预处理逻辑即可覆盖90%以上的异常场景：

@pandas_udf(StringType())
def loc_to_iso3(loc_series: pd.Series) -> pd.Series:
    convert_result = []
    for raw_loc in loc_series:
        loc_str = str(raw_loc).strip()
        # 清洗Row(...)格式的异常字符串
        loc_str = re.sub(r"^Row\([^=]+='(.*?)'\)$", r"\1", loc_str)
        # 提取逗号后段的国家信息，处理「城市,国家」类地址
        if "," in loc_str:
            loc_str = loc_str.split(",")[-1].strip()
        # 执行转换，未匹配值统一标记
        iso3_code = cc.convert(
            names=loc_str, 
            to=OUT_FORMAT, 
            not_found="UNMATCHED", 
            enforce_list=False
        )
        convert_result.append(iso3_code)
    return pd.Series(convert_result)

预处理逻辑覆盖的场景包括：

• 正则清洗Row序列化残留的格式包裹，提取真实地址文本
• 对带逗号的拼接地址，自动提取逗号后的国家段做匹配
• 匹配失败的结果统一标记为UNMATCHED，后续可批量映射

如果仍有少量特殊值无法匹配，只需要收集所有UNMATCHED的取值做一次批量映射即可，不需要逐行硬编码替换：

# 自定义特殊值映射表
CUSTOM_MAPPING = {
    "UNMATCHED": None,
    # 可按需补充特殊映射，例如"Republic of Korea": "KOR"
}
result_df = result_df.replace(CUSTOM_MAPPING, subset=["Location"])

注意：原方案使用collect()拉取全量数据到本地的方式，在数据量超过Driver节点内存时会直接报错，分布式UDF方案按分区批量处理，性能和稳定性更适合生产环境使用。

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