是否存在比.apply()更慢可控的替代方法？地理编码API限流怎么解决？

我刚好遇到过类似的困扰，地理编码API的请求限制确实很磨人，不过有几个靠谱的方案能帮你适配这些规则，下面给你详细拆解：

方法一：利用geopy自带的节流机制（推荐Nominatim用户） #

如果用的是Nominatim这类有官方使用规范的开源API，geopy已经内置了延迟控制，你只需要初始化编码器时明确配置即可：

import pandas as pd
from geopy.geocoders import Nominatim

# 初始化编码器，必填user_agent（否则会被API拒绝），设置1秒延迟（符合Nominatim的每秒1次请求要求）
geolocator = Nominatim(user_agent="your_unique_app_name", delay=1)

def fetch_coordinates(address):
    try:
        location = geolocator.geocode(address)
        return (location.latitude, location.longitude) if location else (None, None)
    except Exception as e:
        print(f"编码失败 {address}: {str(e)}")
        return (None, None)

# 应用到DataFrame并拆分经纬度列
df['coordinates'] = df['address_column'].apply(fetch_coordinates)
df[['latitude', 'longitude']] = pd.DataFrame(df['coordinates'].tolist(), index=df.index)

方法二：手动添加休眠（适配无内置节流的API） #

如果用的是商业API或其他无内置延迟的服务，直接在请求后加入休眠逻辑，给API留足缓冲时间：

import time

def fetch_coords_with_sleep(address):
    try:
        location = geolocator.geocode(address)
        time.sleep(1.5)  # 比API要求的间隔多留0.5秒缓冲，避免触发限制
        return (location.latitude, location.longitude) if location else (None, None)
    except Exception as e:
        print(f"编码失败 {address}: {str(e)}")
        time.sleep(5)  # 出错后延长休眠，防止频繁重试被封禁
        return (None, None)

df['coordinates'] = df['address_column'].apply(fetch_coords_with_sleep)

方法三：分批次处理（适合大数量级数据） #

如果你的DataFrame数据量很大，分批次处理+批量休眠能大幅降低请求密度，避免触发配额上限：

batch_size = 50
coords_results = []

for start_idx in range(0, len(df), batch_size):
    end_idx = start_idx + batch_size
    batch_addresses = df['address_column'].iloc[start_idx:end_idx]
    batch_coords = batch_addresses.apply(fetch_coordinates)
    coords_results.extend(batch_coords)
    print(f"已完成第 {start_idx//batch_size + 1} 批数据处理")
    time.sleep(10)  # 批次间休眠10秒，进一步降低请求频率

df['coordinates'] = coords_results
df[['latitude', 'longitude']] = pd.DataFrame(df['coordinates'].tolist(), index=df.index)

方法四：结合重试库处理临时请求失败 #

用tenacity库实现自动重试逻辑，万一遇到API临时拒绝请求，不用中断整个编码流程：

from tenacity import retry, stop_after_attempt, wait_fixed

# 最多重试3次，每次重试前等待2秒
@retry(stop=stop_after_attempt(3), wait=wait_fixed(2))
def geocode_with_retry(address):
    location = geolocator.geocode(address)
    if not location:
        raise ValueError(f"未找到地址 {address} 的位置信息")
    return (location.latitude, location.longitude)

def fetch_coords_retry(address):
    try:
        result = geocode_with_retry(address)
        time.sleep(1)
        return result
    except Exception as e:
        print(f"最终编码失败 {address}: {str(e)}")
        return (None, None)

df['coordinates'] = df['address_column'].apply(fetch_coords_retry)

额外小建议 #

• 务必仔细阅读目标API的官方文档，明确具体的请求限制（每秒次数、每日配额等），不要凭经验设置延迟
• 缓存已经编码过的地址！可以用字典临时存储，或者把结果导出到CSV，下次直接复用，避免浪费配额
• 如果数据量极大，考虑使用付费API（如Google Maps、OpenCage），这类服务的请求限制更宽松，编码速度也更快

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