Python中Geocoding API地理编码提速方案咨询

原代码处理60个地址耗时约1分钟，对于33万地址的量级来说完全不可行，下面从核心优化、错误处理、数据存储三个维度给出高效实现方案：

一、原代码的核心问题 #

• 同步阻塞请求：每个地址请求后强制sleep 1秒，完全串行处理，浪费大量等待时间
• 错误处理缺失：遇到无效地址、服务器错误等情况会直接崩溃，没有容错机制
• 数据存储不合理：把地址、纬度、经度逐个追加到列表，后续难以直接转换成结构化数据
• 冗余操作：API密钥在循环内重复定义，无意义

二、优化后的实现方案 #

1. 异步并发请求（效率提升核心） #

使用aiohttp实现异步HTTP请求，同时发起多个请求，大幅压缩总耗时。需要控制并发数，避免触发Google API的配额限制（免费版默认每秒5次请求，付费版可根据配额调整）。

2. 指数退避处理配额超限 #

遇到OVER_QUERY_LIMIT状态时，采用指数增长的等待时间重试（比如1秒、2秒、4秒...），而非无限循环等待，提升容错效率。

3. 完善错误处理 #

处理ZERO_RESULTS（地址无匹配）、INVALID_REQUEST（请求格式错误）、SERVER_ERROR（服务器临时错误）等所有可能的状态，避免程序崩溃，同时记录异常地址。

4. 结构化数据存储 #

直接存储每个地址的结果元组，最后转换成pandas.DataFrame，方便按要求输出地址、纬度、经度的结构化结果。

三、完整优化代码 #

import asyncio
import aiohttp
import pandas as pd
from typing import List, Tuple

# 配置参数
GOOGLE_PLACES_API_KEY = 'kkkkkkk'
MAX_CONCURRENT_REQUESTS = 5  # 对应免费版配额，付费版可提高
MAX_RETRIES = 5  # 最大重试次数

async def get_lat_lng(session: aiohttp.ClientSession, address: str) -> Tuple[str, float, float]:
    url = f'https://maps.googleapis.com/maps/api/geocode/json?address={address}&key={GOOGLE_PLACES_API_KEY}'
    retry_count = 0
    
    while retry_count < MAX_RETRIES:
        try:
            async with session.get(url) as response:
                js = await response.json()
                status = js.get('status')
                
                if status == 'OK':
                    location = js['results'][0]['geometry']['location']
                    return address, location['lat'], location['lng']
                elif status == 'OVER_QUERY_LIMIT':
                    # 指数退避
                    await asyncio.sleep(2 ** retry_count)
                    retry_count += 1
                elif status in ['ZERO_RESULTS', 'INVALID_REQUEST']:
                    # 无匹配地址或请求无效，返回空值标记
                    return address, None, None
                else:
                    # 其他错误，重试
                    await asyncio.sleep(1)
                    retry_count += 1
        except Exception as e:
            print(f"请求地址 {address} 时出错: {str(e)}")
            await asyncio.sleep(1)
            retry_count += 1
    
    # 超过最大重试次数，返回空值
    return address, None, None

async def batch_geocode(addresses: List[str]) -> List[Tuple[str, float, float]]:
    async with aiohttp.ClientSession() as session:
        # 控制并发数
        semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
        async def bounded_get(address):
            async with semaphore:
                return await get_lat_lng(session, address)
        
        tasks = [bounded_get(addr) for addr in addresses]
        results = await asyncio.gather(*tasks)
        return results

if __name__ == '__main__':
    addressList = ['xxx', 'xx1']  # 实际33万地址
    # 运行异步任务
    results = asyncio.run(batch_geocode(addressList))
    
    # 转换成DataFrame并过滤无效结果
    df = pd.DataFrame(results, columns=['address', 'latitude', 'longitude'])
    df = df.dropna(subset=['latitude', 'longitude'])  # 可选：移除解析失败的地址
    
    # 按要求输出（可保存为CSV、Excel等）
    print(df)
    # df.to_csv('geocode_results.csv', index=False)

四、额外效率建议 #

• 缓存已解析结果：如果地址列表存在重复，先通过字典缓存已解析的坐标，避免重复请求API
• 分批处理大列表：33万地址可分成每批次1万条处理，避免内存占用过高，同时便于断点续跑
• 升级API配额：如果预算允许，升级到付费版Google Geocoding API，获得更高的请求频率限制，进一步提升效率

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