如何提升Python调用Mapbox获取经纬度的速度？

一、验证你的字典构建逻辑 #

你的字典构建逻辑是正确的，不过可以简化：不需要把coords转成pd.Series，直接用dict(zip(addresses, coords))就能得到地址到坐标的映射字典。

如果要快速验证，不用跑全量数据，先取少量地址测试：

# 取前5个唯一地址测试
test_addresses = df['Address'].str.lower().unique()[:5]
test_coords = []
for addr in test_addresses:
    response = geocoder.forward(addr)
    # 增加异常处理，避免地址解析失败导致索引报错
    if response.geojson()["features"]:
        test_coords.append(response.geojson()["features"][0]["center"])
    else:
        test_coords.append(None)  # 解析失败标记为None

test_dict = dict(zip(test_addresses, test_coords))
print(test_dict)

这样能快速确认字典是否正确映射，同时还能处理地址解析失败的极端情况。

二、优化50万行数据的地理编码性能 #

原代码仅处理了唯一地址，但同步循环请求依然缓慢，核心原因是同步HTTP请求的串行等待。下面是两种高效解决方案：

方案1：使用Mapbox官方批量地理编码API #

Mapbox提供Batch Geocoding API，一次可提交最多100个地址，大幅减少请求次数，是官方推荐的批量处理方式。

示例代码：

import requests

MAPBOX_TOKEN = "---"
BATCH_URL = "https://api.mapbox.com/geocoding/v5/mapbox.places-batch"

addresses = df['Address'].str.lower().unique()
# 拆分批次，每100个地址一组
batches = [addresses[i:i+100] for i in range(0, len(addresses), 100)]

addr_coords = {}

for batch in batches:
    # 构造批量请求体：每个地址带唯一id（用批次内索引）
    batch_body = [{"id": idx, "text": addr} for idx, addr in enumerate(batch)]
    params = {"access_token": MAPBOX_TOKEN}
    
    response = requests.post(BATCH_URL, json=batch_body, params=params)
    response_data = response.json()
    
    # 解析响应，匹配地址与坐标
    for result in response_data["batch"]:
        addr = batch[result["id"]]
        if result["features"]:
            addr_coords[addr] = result["features"][0]["center"]
        else:
            addr_coords[addr] = None

# 映射回原DataFrame
df['Coords'] = df['Address'].str.lower().map(addr_coords)

注意：需提前查看你的Mapbox账户批量请求速率限制，避免触发限流。

方案2：异步并发请求 #

如果不想用批量API，可使用aiohttp异步库并发发起请求，同时处理多个地址解析，比同步循环效率提升数倍。

示例代码（带并发数限制）：

import aiohttp
import asyncio

MAPBOX_TOKEN = "---"
GEOCODE_URL = "https://api.mapbox.com/geocoding/v5/mapbox.places/{address}.json?access_token={token}"

async def geocode_address(session, address, semaphore):
    async with semaphore:
        url = GEOCODE_URL.format(address=address.replace(" ", "%20"), token=MAPBOX_TOKEN)
        try:
            async with session.get(url) as response:
                data = await response.json()
                if data.get("features"):
                    return (address, data["features"][0]["center"])
                else:
                    return (address, None)
        except Exception as e:
            print(f"解析地址{address}失败: {str(e)}")
            return (address, None)

async def main(addresses):
    # 限制同时发起10个请求，避免触发API速率限制
    semaphore = asyncio.Semaphore(10)
    async with aiohttp.ClientSession() as session:
        tasks = [geocode_address(session, addr, semaphore) for addr in addresses]
        results = await asyncio.gather(*tasks)
        return dict(results)

# 运行异步任务
addresses = df['Address'].str.lower().unique()
addr_coords = asyncio.run(main(addresses))

# 映射回原DataFrame
df['Coords'] = df['Address'].str.lower().map(addr_coords)

额外优化建议 #

• 本地缓存：将地址-坐标映射保存到JSON文件或数据库，后续任务可直接复用，避免重复请求API。
• 异常兜底：在所有请求逻辑中添加try-except块，处理网络超时、API返回错误等情况，避免程序崩溃。

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