如何优化Pandas+ArcGIS REST API批量地址地理编码的执行效率？

原代码处理1600条地址耗时30分钟，核心瓶颈是单地址串行请求+逐行写入文件。以下提供两种优化方案，优先推荐批量接口方案，效率提升最显著：

方案一：使用ArcGIS批量地理编码接口（最优） #

ArcGIS Geocode服务提供geocodeAddresses批量接口，支持单次提交最多1000条地址（具体以服务限制为准），1600条仅需2次请求即可完成，大幅降低网络开销。

核心优化点 #

• 将1600次单地址请求压缩为2次批量请求
• 批量收集结果后一次性写入CSV，减少文件IO操作
• 增加异常捕获，避免单个地址失败中断整个流程
• 通过OBJECTID关联原始地址与匹配结果，确保数据对应关系

代码示例 #

import pandas as pd
import requests
import csv

# 批量地理编码接口地址
BATCH_URL = "https://geocode.arcgis.com/arcgis/rest/services/World/GeocodeServer/geocodeAddresses"
BASE_PARAMS = {
    'f': 'pjson',
    'forStorage': 'false',
    'Country': 'CA',
    'City': 'Oakville',
    'Region': 'ON'
}
EXCEL_PATH = "你的Excel文件路径"
CSV_SAVE_PATH = "结果CSV路径"
MISSING_FILE_PATH = r"C:\Users\PC\Documents\missing.txt"

def prepare_batch_addresses(addresses):
    """将地址列表转换为批量接口要求的格式"""
    features = []
    for idx, addr in enumerate(addresses):
        features.append({
            "attributes": {
                "OBJECTID": idx + 1,
                "Address": addr
            }
        })
    return {"records": features}

def geocode_batch(addresses):
    """发起批量地理编码请求"""
    batch_data = prepare_batch_addresses(addresses)
    response = requests.post(BATCH_URL, params=BASE_PARAMS, json=batch_data)
    response.raise_for_status()  # 捕获HTTP错误
    return response.json()

def process_batch_results(results, original_addresses):
    """处理批量返回结果，筛选有效记录并收集未匹配地址"""
    valid_records = []
    missing = []
    result_dict = {res['attributes']['OBJECTID']: res for res in results['locations']}
    
    for idx, addr in enumerate(original_addresses, 1):
        res = result_dict.get(idx)
        if not res:
            missing.append(addr)
            continue
        score = res['attributes']['Score']
        if score >= 95:
            valid_records.append([
                res['attributes']['Match_addr'],
                res['attributes']['Addr_type'],
                res['location']['y'],
                res['location']['x']
            ])
        else:
            missing.append(addr)
    return valid_records, missing

def main():
    # 读取Excel数据
    df = pd.read_excel(EXCEL_PATH, sheet_name="line_pts", usecols="A", dtype=str)
    addresses = df["DATA"].tolist()
    
    # 分批次处理（每1000条一批）
    batch_size = 1000
    all_valid = []
    all_missing = []
    
    for i in range(0, len(addresses), batch_size):
        batch = addresses[i:i+batch_size]
        try:
            batch_result = geocode_batch(batch)
            valid, missing = process_batch_results(batch_result, batch)
            all_valid.extend(valid)
            all_missing.extend(missing)
            print(f"完成第{i//batch_size +1}批处理，共{len(batch)}条")
        except Exception as e:
            print(f"第{i//batch_size +1}批处理失败: {str(e)}")
            all_missing.extend(batch)
    
    # 写入CSV结果
    with open(CSV_SAVE_PATH, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f, dialect='excel')
        writer.writerow(["Address", "Type", "Lat", "Long"])
        writer.writerows(all_valid)
    
    # 写入未匹配地址
    with open(MISSING_FILE_PATH, "w", encoding="utf-8") as f:
        f.write('Missed/Non-located places\n')
        for addr in all_missing:
            f.write(f"{addr}\n")
    
    missing_rate = (len(all_missing)/len(addresses))*100
    print(f"{CSV_SAVE_PATH} 写入完成。未匹配数据占比: {missing_rate:.2f}%")

if __name__ == "__main__":
    main()

方案二：多线程并行单地址请求 #

若无法使用批量接口（如权限限制），可通过线程池并行发起请求，利用网络等待时间处理其他任务，提升整体效率。

核心优化点 #

• 使用ThreadPoolExecutor实现IO密集型任务并行
• 每次请求生成独立参数字典，避免全局变量竞争
• 批量收集结果后写入文件，减少IO开销
• 增加超时与异常捕获，提升鲁棒性

代码示例 #

import pandas as pd
import requests
import csv
from concurrent.futures import ThreadPoolExecutor, as_completed

URL = "https://geocode.arcgis.com/arcgis/rest/services/World/GeocodeServer/findAddressCandidates?"
BASE_PARAMS = {'f':'pjson','outFields':'Addr_type','forStorage':'false','City':'Oakville','Region':'ON'}
EXCEL_PATH = "你的Excel文件路径"
CSV_SAVE_PATH = "结果CSV路径"
MISSING_FILE_PATH = r"C:\Users\PC\Documents\missing.txt"

def geocode_single(address):
    """单地址地理编码请求，返回有效结果或未匹配地址"""
    params = BASE_PARAMS.copy()
    params['Address'] = address
    try:
        response = requests.get(URL, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        candidates = data.get('candidates', [])
        if not candidates:
            return None, address
        data_dict = candidates[0]
        if data_dict.get('score') >= 95:
            return [
                data_dict.get('address'),
                data_dict.get('attributes').get('Addr_type'),
                data_dict.get('location').get('y'),
                data_dict.get('location').get('x')
            ], None
        else:
            return None, address
    except Exception as e:
        print(f"地址 {address} 请求失败: {str(e)}")
        return None, address

def main():
    df = pd.read_excel(EXCEL_PATH, sheet_name="line_pts", usecols="A", dtype=str)
    addresses = df["DATA"].tolist()
    
    all_valid = []
    all_missing = []
    
    # 线程池大小建议10-20，避免触发ArcGIS请求频率限制
    with ThreadPoolExecutor(max_workers=15) as executor:
        futures = {executor.submit(geocode_single, addr): addr for addr in addresses}
        for future in as_completed(futures):
            result, missing_addr = future.result()
            if result:
                all_valid.append(result)
            if missing_addr:
                all_missing.append(missing_addr)
    
    # 写入CSV结果
    with open(CSV_SAVE_PATH, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f, dialect='excel')
        writer.writerow(["Address", "Type", "Lat", "Long"])
        writer.writerows(all_valid)
    
    # 写入未匹配地址
    with open(MISSING_FILE_PATH, "w", encoding="utf-8") as f:
        f.write('Missed/Non-located places\n')
        for addr in all_missing:
            f.write(f"{addr}\n")
    
    missing_rate = (len(all_missing)/len(addresses))*100
    print(f"{CSV_SAVE_PATH} 写入完成。未匹配数据占比: {missing_rate:.2f}%")

if __name__ == "__main__":
    main()

额外优化建议 #

• 请求频率控制：ArcGIS免费版存在请求频率限制，批量或并行请求时避免超过阈值，必要时可加入短延时
• 地址预处理：提前清洗地址数据（如去除多余空格、统一格式），提升匹配成功率与速度
• 失败重试：对网络波动导致的失败请求，可加入自动重试机制（如使用tenacity库）

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