50万行DataFrame分批执行df.merge、df.apply及谷歌地图API请求方案

核心思路 #

将50万行的大DataFrame拆分为100行一组的小批次，逐个处理每一批（包含merge、apply、谷歌API调用逻辑），处理完成后直接追加写入目标CSV文件，同时给API请求添加延迟，避免短时间请求量超限导致崩溃。

1. 拆分大DataFrame为小批次 #

用numpy.array_split自动拆分，确保最后一批不足100行也能被处理：

import pandas as pd
import numpy as np

# 读取原始大文件
df = pd.read_csv("your_large_input.csv")
batch_size = 100
# 拆分批次，len(df)//batch_size +1 保证所有行都被覆盖
batches = np.array_split(df, len(df) // batch_size + 1)

2. 封装单批次处理逻辑 #

把merge、API调用、数据处理的逻辑打包成函数，同时给API请求加延迟：

import time
import googlemaps

# 初始化谷歌地图API客户端（替换成你的密钥）
gmaps_client = googlemaps.Client(key="your_google_api_key")

def process_single_batch(batch_df):
    # 执行你的merge操作（替换成实际的关联逻辑）
    merged_batch = batch_df.merge(your_other_dataframe, on="join_column", how="left")
    
    # 定义行处理函数，包含API调用和延迟
    def process_row(row):
        try:
            # 调用谷歌API（替换成你的实际API请求逻辑）
            api_response = gmaps_client.geocode(row["target_address"])
            time.sleep(0.1)  # 每次请求后延迟0.1秒，可根据API配额调整
            
            # 解析API结果（示例：提取经纬度）
            if api_response:
                row["latitude"] = api_response[0]["geometry"]["location"]["lat"]
                row["longitude"] = api_response[0]["geometry"]["location"]["lng"]
            else:
                row["latitude"] = None
                row["longitude"] = None
        except Exception as e:
            print(f"处理行{row.name}出错: {str(e)}")
            row["latitude"] = None
            row["longitude"] = None
        return row
    
    # 批量处理当前批次的行
    processed_batch = merged_batch.apply(process_row, axis=1)
    return processed_batch

3. 分批处理并追加写入CSV #

第一次写入时保留表头，后续批次追加时跳过表头，避免重复：

is_first_batch = True
total_batches = len(batches)

for batch_idx, batch in enumerate(batches):
    print(f"处理第 {batch_idx+1}/{total_batches} 批数据")
    processed_data = process_single_batch(batch)
    
    # 写入CSV：第一次写表头，之后追加不写
    processed_data.to_csv(
        "final_output.csv",
        mode="a",
        header=is_first_batch,
        index=False
    )
    is_first_batch = False
    
    # 可选：每处理10批增加一次长延迟，进一步降低API请求压力
    if (batch_idx + 1) % 10 == 0:
        time.sleep(5)

额外优化建议 #

• 优先用批量API：如果谷歌地图API支持批量请求（比如批量地理编码），直接用批量接口替代逐行调用，大幅减少请求次数
• 进度可视化：用tqdm库给循环加进度条，更直观跟踪处理进度：

  from tqdm import tqdm
  for batch_idx, batch in enumerate(tqdm(batches)):
      # 处理逻辑同上
  

• 断点续传：如果担心中途崩溃，可以把已处理的批次号记录到文件，重启时从断点继续处理

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