PySpark DataFrame基于完整单词匹配关联提取城市信息

问题背景 #

我有两个PySpark DataFrame：

• pubmed_clean：高校地址数据集，含有人工输入的完整地址字符串，需要从中提取城市、国家信息
• worldcities：全球城市主数据集，包含城市、国家的标准列表

之前用contains()做关联匹配时，会出现子字符串误匹配的问题（比如把"York"匹配到"New York"的子串）。现在希望通过正则表达式实现完整单词级别的精确匹配，避免拆分地址列的繁琐操作。

现有代码如下：

import pyspark.sql.functions as F
from pyspark.sql.window import Window

def pubmed2(pubmed_clean, worldcities):
    df = pubmed_clean.selectExpr("AffiliationInfo as address").filter((F.col('address').isNotNull())).distinct()
    w = Window().orderBy(F.lit('A'))
    df = df.withColumn("row_num", F.row_number().over(w))
    df= df.filter(F.col("row_num")<=100)
    df = df.select([F.lower(F.col(c)).alias(c) for c in df.columns])
    world_cities=worldcities.select([F.lower(F.col(c)).alias(c) for c in worldcities.columns])
    df = udf(df, world_cities)
    final = df.dropDuplicates()
    return final

def udf(df,world_cities):
    world_countries = (world_cities.selectExpr("country").distinct())
    df = (df.join(world_countries, df.address.contains(world_countries.country), how='left'))
    world_countries = (world_cities.selectExpr("city").distinct())
    df = df.join(world_countries,df.address.contains(world_countries.city), how='left')
    
    return df

示例地址数据：

1. 巴塞尔大学初级卫生保健中心，巴塞尔州立医院，莱茵街26号，4410，利斯塔，瑞士。
2. 首尔国立大学医院江南医疗系统中心家庭医学科，首尔，06236，大韩民国。
3. 沙希德·贝赫什蒂医科大学护理与助产学学院内外科护理系，德黑兰，伊朗。
4. 热基蒂尼奥尼亚和穆库里谷联邦大学护理系，迪亚曼蒂纳，米纳斯吉拉斯州，巴西。米纳斯吉拉斯联邦大学生物工程实验室，贝洛奥里藏特，米纳斯吉拉斯州，巴西。
5. 诺丁汉大学生物科学学院食品、营养与饮食学系感官科学中心，萨顿博宁顿校区，莱斯特郡，LE12 5RD，英国。
6. 密歇根大学医学院，安娜堡，密歇根州，美国。

解决方案 #

核心改动是用PySpark的rlike函数结合正则表达式，实现完整单词的精确匹配，替代原有的contains方法。

修改后的完整代码 #

import pyspark.sql.functions as F
from pyspark.sql.window import Window

def pubmed2(pubmed_clean, worldcities):
    df = pubmed_clean.selectExpr("AffiliationInfo as address").filter(F.col('address').isNotNull()).distinct()
    # 取前100条数据测试
    w = Window().orderBy(F.lit('A'))
    df = df.withColumn("row_num", F.row_number().over(w)).filter(F.col("row_num") <= 100)
    # 统一转为小写，避免大小写匹配问题
    df = df.select([F.lower(F.col(c)).alias(c) for c in df.columns])
    world_cities = worldcities.select([F.lower(F.col(c)).alias(c) for c in worldcities.columns])
    
    df = match_full_word(df, world_cities)
    final = df.dropDuplicates()
    return final

def match_full_word(df, world_cities):
    # 匹配国家：完整单词匹配
    country_list = world_cities.selectExpr("country").distinct()
    # 构建正则模式：匹配独立的国家名，前后是非单词/连字符（处理带连字符的国家名）
    df = df.join(
        country_list,
        df.address.rlike(F.concat(F.lit(r'(?<![\w-])'), country_list.country, F.lit(r'(?![\w-])'))),
        how='left'
    ).withColumnRenamed("country", "matched_country")
    
    # 匹配城市：完整单词匹配
    city_list = world_cities.selectExpr("city").distinct()
    df = df.join(
        city_list,
        df.address.rlike(F.concat(F.lit(r'(?<![\w-])'), city_list.city, F.lit(r'(?![\w-])'))),
        how='left'
    ).withColumnRenamed("city", "matched_city")
    
    return df

关键改动说明 #

1. 正则表达式逻辑：

   • (?<![\w-])：反向零宽断言，确保匹配的单词前面不是字母、数字或连字符（避免子字符串匹配，比如把"York"匹配到"New York"）
   • (?![\w-])：正向零宽断言，确保匹配的单词后面不是字母、数字或连字符
   • 用concat拼接正则模板和目标字段，实现动态匹配

2. 字段重命名：

   • 把匹配后的country和city重命名为matched_country、matched_city，避免字段冲突

3. 避免大小写问题：

   • 提前把两个数据集的所有字段转为小写，确保匹配不受大小写影响

4. 处理多匹配情况：

   • 一个地址可能包含多个城市/国家（比如示例4有两个城市），如果需要保留所有匹配结果，直接保留当前逻辑；如果需要唯一匹配，可以添加额外筛选逻辑（比如取最后一个匹配项、最长匹配项等）

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