PySpark调用geopy的Nominatim反地理编码报AttributeError如何解决

问题根因 #

报错是因为Nominatim定位器实例定义在UDF函数外部，PySpark需要将UDF依赖的所有对象序列化后分发到各工作节点执行，而geopy内部的HTTP请求适配器实例不支持Spark的序列化机制，传输到工作节点后丢失属性抛出异常。本地pandas运行时不需要跨节点序列化传输对象，因此运行正常。

解决方法 #

将Nominatim实例的初始化逻辑移到getCity函数内部，让每个工作节点执行函数时在本地生成独立的定位器实例，避开序列化问题。

修改后代码 #

from geopy.geocoders import Nominatim
import pyspark.sql.functions as F
from pyspark.sql.types import StringType

def getCity(coords):
    # 定位器初始化移到函数内部，避免序列化问题
    locator = Nominatim(user_agent="myGeocoder")
    loc = locator.reverse(coords, language="en")
    try:
        city = loc.raw["address"]["city"]
    except:
        city = '' if loc is None else loc.raw["address"].get('state', '') # 未查询到城市则返回州，否则返回空字符串
    return city

UDF_getCity  = F.udf(getCity, StringType())
cityDF_spark = cityDF_spark.select('geom', UDF_getCity('geom').alias('CITY'))
cityDF_spark.show()

额外注意事项 #

• 若数据量较大，需注意Nominatim公共接口的请求频率限制，高频请求可能会被封禁，建议批量请求时加适当延迟，或自建本地Nominatim服务提升性能和稳定性

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