如何用Python多进程高效向Woosmap API发送50万地理编码请求

针对50万条地址的地理编码需求，单线程请求效率极低，使用Python多进程可以充分利用多核CPU并发发送请求，大幅缩短总耗时。以下是具体实现方案和代码示例：

核心思路 #

1. 封装独立的地理编码函数，处理单条地址的请求与解析
2. 使用multiprocessing.Pool实现并发请求，控制进程数匹配API速率限制
3. 完善错误处理与重试机制，避免数据丢失
4. 将结果合并回原DataFrame并保存

完整代码示例 #

1. 导入依赖库 #

import pandas as pd
import requests
from multiprocessing import Pool
from urllib.parse import quote_plus
from tenacity import retry, stop_after_attempt, wait_exponential  # 可选，用于重试逻辑

2. 定义地理编码函数 #

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def geocode_address(args):
    address, country, api_key = args
    base_url = "https://api.woosmap.com/localities/geocode"
    
    # URL编码地址，避免特殊字符导致请求失败
    encoded_address = quote_plus(str(address))
    params = {
        "address": encoded_address,
        "components": f"country:{country}",
        "key": api_key
    }
    headers = {
        'Referer': 'http://localhost'
    }

    try:
        response = requests.get(base_url, params=params, headers=headers, timeout=15)
        response.raise_for_status()  # 捕获HTTP状态码错误（如403、500）
        data = response.json()
        
        if data.get('results') and len(data['results']) > 0:
            location = data['results'][0]['geometry']['location']
            return (location['lat'], location['lng'], None)
        else:
            return (None, None, "No matching results found")
    except requests.exceptions.RequestException as e:
        return (None, None, f"Request failed: {str(e)}")

3. 多进程批量处理 #

# 加载你的地址数据集（假设包含address和country列）
df = pd.read_csv("your_address_dataset.csv")
API_KEY = "YOUR_PUBLIC_API_KEY"

# 构造参数列表：每个元素对应一条地址的请求参数
args_list = list(zip(df['address'], df['country'], [API_KEY] * len(df)))

# 设置并发进程数：参考Woosmap API的速率限制（如QPS=100则设为100左右）
# 避免进程数过高导致API封禁或网络拥堵
process_count = 80

# 启动多进程池处理
with Pool(processes=process_count) as pool:
    # 用map批量执行，结果顺序与输入一致
    geocode_results = pool.map(geocode_address, args_list)

# 将结果拆分到原DataFrame的新列
df['latitude'], df['longitude'], df['geocode_error'] = zip(*geocode_results)

# 保存最终结果（建议保存为CSV或Parquet，避免数据丢失）
df.to_csv("geocoded_addresses.csv", index=False)

关键注意事项 #

• API速率限制：必须严格遵守Woosmap的请求速率限制（查看官方文档确认QPS上限），进程数不要超过限制值，否则会触发API封禁。
• 重试机制：使用tenacity库的重试装饰器可以自动处理网络波动导致的请求失败，避免重复手动处理失败数据。
• 数据分批：如果50万条数据一次性处理内存压力过大，可以拆分多个批次（如每10万条处理一次），分批保存结果。
• 错误排查：geocode_error列会记录所有失败原因，处理完后可以筛选出错误行单独重新处理。
• URL编码：必须对地址进行URL编码，否则包含特殊字符（如空格、逗号）的地址会导致请求参数无效。

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