PySpark UDF调用谷歌地理编码API返回[Ljava.lang.Object;@问题求助

问题根源 #

你的UDF返回的是Python元组(lat, lng)，但PySpark无法自动识别这种复合类型，必须显式指定匹配的Spark数据类型。之前尝试的FloatType（单个数值）、StringType（字符串）都和返回值类型不匹配，导致出现[Ljava.lang.Object;@xxxx（Java数组的字符串表示）或NULL。

解决方案 #

1. 定义正确的返回数据类型

根据需求选择两种类型之一：

• 结构体类型（StructType）：返回带字段名的经纬度，便于后续提取
• 数组类型（ArrayType）：返回经纬度数组

代码示例：

# 导入Spark数据类型
from pyspark.sql.types import StructType, StructField, FloatType, ArrayType

# 选项1：结构体类型（推荐，语义清晰）
coord_struct_schema = StructType([
    StructField("latitude", FloatType(), nullable=True),
    StructField("longitude", FloatType(), nullable=True)
])

# 选项2：数组类型
coord_array_schema = ArrayType(FloatType(), containsNull=True)

2. 修改UDF与函数逻辑

优化原函数，处理异常情况（如地址为空、API请求失败、无解析结果），并确保返回值与定义的schema匹配：

import requests

def getLocCoordinates(property_address):
    # 处理空地址或非字符串地址
    if not property_address or not isinstance(property_address, str):
        return None
    
    url = "https://maps.googleapis.com/maps/api/geocode/json"
    querystring = {"address": property_address.strip(), "key": "THE_API_KEY"}
    
    try:
        # 发送请求并处理HTTP错误
        response = requests.get(url, params=querystring)
        response.raise_for_status()
        response_json = response.json()
        
        # 检查是否有解析结果
        if not response_json.get("results"):
            return None
        
        # 取第一个解析结果的经纬度
        first_result = response_json["results"][0]
        lat = first_result["geometry"]["location"]["lat"]
        lng = first_result["geometry"]["location"]["lng"]
        
        # 对应结构体类型返回字典，对应数组类型返回元组/列表
        # 结构体类型返回：{"latitude": lat, "longitude": lng}
        # 数组类型返回：(lat, lng)
        return {"latitude": lat, "longitude": lng}
    
    except Exception as e:
        # 打印错误便于调试，生产环境可改为日志记录
        print(f"解析地址 {property_address} 失败: {str(e)}")
        return None

3. 重新创建UDF并应用到DataFrame

根据选择的schema创建UDF：

# 使用结构体类型的UDF
getCoorsUDF = udf(getLocCoordinates, returnType=coord_struct_schema)

# 或者使用数组类型的UDF
# getCoorsUDF = udf(getLocCoordinates, returnType=coord_array_schema)

# 生成新列
df = df.withColumn("AddressCoordinates", getCoorsUDF(col("FullAddress")))

后续提取经纬度 #

如果用结构体类型，可以直接提取字段：

df = df.withColumn("Latitude", col("AddressCoordinates.latitude")) \
       .withColumn("Longitude", col("AddressCoordinates.longitude"))

如果用数组类型，通过索引提取：

df = df.withColumn("Latitude", col("AddressCoordinates")[0]) \
       .withColumn("Longitude", col("AddressCoordinates")[1])

额外注意事项 #

• 谷歌地理编码API有调用限额，建议对地址去重后再调用，避免浪费配额
• 可以添加请求延迟（如time.sleep(0.1)），避免触发API限流
• 生产环境建议将API密钥放在环境变量中，不要硬编码在代码里

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