如何显著提升 Python 地理编码（Geocoding）批量处理速度

本文介绍通过多进程并发 + 请求节流策略优化 Google Geocoding API 批量调用性能，将 33 万地址的地理编码耗时从单线程分钟级降至小时级可控范围，兼顾效率与 API 合规性。

本文介绍通过多进程并发 + 请求节流策略优化 Google Geocoding API 批量调用性能，将 33 万地址的地理编码耗时从单线程分钟级降至小时级可控范围，兼顾效率与 API 合规性。

地理编码（Geocoding）是将地址文本转换为经纬度坐标的常见任务，但在处理大规模地址数据（如 33 万条）时，原始单线程串行请求方式极易成为性能瓶颈——不仅因每请求强制 time.sleep(1) 导致吞吐极低，还缺乏错误重试、状态校验与资源复用机制。

核心优化思路：并行化 + 节流控制 + 结构化输出
Google Maps Platform 对 Geocoding API 实施严格的配额与速率限制（默认 50 QPS，每日免费额度 40,000 次）。盲目增加并发数反而触发 OVER_QUERY_LIMIT，引发大量失败和重试，最终拉低整体吞吐。因此，合理控制并发请求数 + 每进程匀速发送 + 全局错误恢复才是高效稳定的关键。

以下是一个生产就绪的优化实现示例（基于 concurrent.futures.ProcessPoolExecutor）：

import requests
import json
import pandas as pd
from concurrent.futures import ProcessPoolExecutor, as_completed
from time import sleep
import logging

# 配置日志（便于调试与监控）
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

GOOGLE_API_KEY = "kkkkkkk"
GEOCODE_URL = "https://maps.googleapis.com/maps/api/geocode/json"

def geocode_single(address: str) -> dict:
    """单地址地理编码，含重试、状态校验与结构化返回"""
    max_retries = 3
    for attempt in range(max_retries):
        try:
            params = {"address": address, "key": GOOGLE_API_KEY}
            res = requests.get(GEOCODE_URL, params=params, timeout=10)
            res.raise_for_status()
            data = res.json()

            if data["status"] == "OK":
                loc = data["results"][0]["geometry"]["location"]
                return {
                    "address": address,
                    "latitude": loc["lat"],
                    "longitude": loc["lng"],
                    "status": "success"
                }
            elif data["status"] == "ZERO_RESULTS":
                return {"address": address, "latitude": None, "longitude": None, "status": "no_result"}
            elif data["status"] == "OVER_QUERY_LIMIT":
                sleep(1.5)  # 遇限退避略长于1秒，避免雪崩
                continue
            else:
                return {"address": address, "latitude": None, "longitude": None, "status": f"api_error:{data['status']}"}
        except Exception as e:
            logger.warning(f"Geocoding failed for '{address}' (attempt {attempt+1}): {e}")
            if attempt < max_retries - 1:
                sleep(2 ** attempt)  # 指数退避
            else:
                return {"address": address, "latitude": None, "longitude": None, "status": f"exception:{str(e)}"}

    return {"address": address, "latitude": None, "longitude": None, "status": "max_retries_exceeded"}

def batch_geocode(addresses: list, max_workers: int = 5) -> pd.DataFrame:
    """并行批量地理编码，返回结构化 DataFrame"""
    results = []
    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        future_to_addr = {executor.submit(geocode_single, addr): addr for addr in addresses}

        # 按完成顺序收集结果（非提交顺序）
        for future in as_completed(future_to_addr):
            result = future.result()
            results.append(result)

    # 按原始地址顺序排序（可选，若需严格保序）
    # results.sort(key=lambda x: addresses.index(x["address"]))

    return pd.DataFrame(results)

# 使用示例
if __name__ == "__main__":
    address_list = ["北京市朝阳区建国路1号", "上海市浦东新区世纪大道100号", "广州市天河区珠江新城"]  # 替换为你的 33 万地址

    # 推荐：5–10 个 worker（取决于 API 配额与网络延迟，避免超限）
    df_result = batch_geocode(address_list, max_workers=8)
    print(df_result[["address", "latitude", "longitude", "status"]])

    # 保存结果
    df_result.to_csv("geocoded_results.csv", index=False, encoding="utf-8-sig")

✅ 关键优势说明：

• 可控并发：max_workers 参数直接控制并发请求数（建议 5–10），配合每个进程内部自然节流，整体速率稳定在安全区间；
• 健壮错误处理：自动重试 OVER_QUERY_LIMIT、超时、网络异常，并记录失败原因；
• 结构化输出：返回 pandas.DataFrame，字段明确包含 address, latitude, longitude, status，便于后续清洗与分析；
• 符合 Google 最佳实践：遵守 Polite Use 原则，避免因激进请求导致 IP 限流或配额冻结。

⚠️ 注意事项：

• 若地址量极大（如 33 万），建议分批执行（例如每批 5,000 条），并加入检查点保存机制，防止中断后全量重跑；
• 生产环境务必使用环境变量管理 GOOGLE_API_KEY，切勿硬编码；
• 对于高精度需求，可启用 components 参数（如限定国家/城市）提升匹配率；
• 考虑缓存已成功解析的地址（如用 functools.lru_cache 或 Redis），避免重复请求。

综上，通过合理引入多进程并行与精细化的请求调度，地理编码任务可在保障 API 合规的前提下实现数量级性能提升——33 万地址在 8 并发、稳定 6–8 QPS 下通常可在 10–12 小时内完成，远优于原始单线程方案的数天耗时。