使用PySpark差异化处理每行NULL值 调用API补全经纬度空值

整体实现逻辑 #

你可以通过地理编码服务将结构化地址转换为经纬度坐标，按以下流程操作即可：

• 第一步：选择合适的地理编码服务，申请对应的调用密钥，确认服务的请求频率限制、配额规则。
• 第二步：读取你的业务数据表，筛选出Latitude、Longitude字段为空的记录，避免重复查询已有坐标的数据。
• 第三步：遍历每一条待处理的记录，建议将Address、City、Country三个字段拼接为完整地址传入API，仅传入城市名称只能获取城市级的粗略坐标，精度极差。
• 第四步：解析API返回结果，提取经纬度数值回写到对应记录的字段中。
• 第五步：将补全后的数据回写到原存储（数据库/CSV文件等）。

Python实现示例 #

以下是基于Pandas+Requests的示例代码，可根据你实际使用的地理编码服务调整请求参数和解析逻辑：

import pandas as pd
import requests
import time

# 读取数据表，若为数据库表可通过sqlalchemy等库直接读取
df = pd.read_csv("你的数据表文件路径.csv")
# 筛选经纬度为空的待处理行
to_process_rows = df[df["Latitude"].isna() | df["Longitude"].isna()]
# 地理编码服务调用密钥
API_KEY = "你申请的服务密钥"
# 地理编码服务接口地址（替换为你实际使用的服务地址）
GEO_API_URL = "地理编码服务接口请求地址"

for idx, row in to_process_rows.iterrows():
    # 拼接完整地址，提升坐标精度
    full_addr = f"{row['Address']}, {row['City']}, {row['Country']}"
    req_params = {
        "address": full_addr,
        "key": API_KEY
    }
    try:
        resp = requests.get(GEO_API_URL, params=req_params, timeout=10)
        resp_json = resp.json()
        # 以下解析逻辑需匹配你所用API的返回结构
        if resp_json.get("status") == "OK":
            loc = resp_json["results"][0]["geometry"]["location"]
            df.at[idx, "Latitude"] = loc["lat"]
            df.at[idx, "Longitude"] = loc["lng"]
            print(f"处理成功：{full_addr}")
        else:
            print(f"处理失败：{full_addr}，错误原因：{resp_json.get('error_message','未知错误')}")
        # 控制请求频率，避免触发限流
        time.sleep(1)
    except Exception as e:
        print(f"请求异常：{full_addr}，异常信息：{str(e)}")

# 处理完成后保存结果，若为数据库表可直接执行update操作回写
df.to_csv("补全经纬度后的结果表.csv", index=False)

注意事项 #

• 若仅需要城市级的粗略坐标，可直接将full_addr替换为f"{row['City']}, {row['Country']}"即可。
• 建议单独记录所有查询失败的地址，后续可人工校验补全。
• 若数据量较大，可使用批量查询接口降低请求次数，提升处理效率。

内容的提问来源于stack exchange，提问作者Rustam-Z