高效处理大型DataFrame：Pandas分批操作与外部API请求管理

import pandas as pd
import numpy as np
import time
import os

# 创建一个大型模拟DataFrame
data_size = 500000 # 50万行数据
df = pd.DataFrame({
    'id': range(data_size),
    'value1': np.random.rand(data_size) * 100,
    'value2': np.random.randint(1, 1000, data_size),
    'address': [f"模拟地址 {i}, 城市A, 国家B" for i in range(data_size)] # 模拟地址信息
})

print(f"原始DataFrame大小: {len(df)} 行")

batch_size = 1000 # 每批处理1000行
df['batch_num'] = df.index // batch_size

# 打印批次信息
print(f"数据将被分割成 {df['batch_num'].nunique()} 个批次，每批 {batch_size} 行。")

# 模拟外部API调用函数
def get_coordinates_from_address(address):
    """
    模拟一个外部API调用，根据地址获取经纬度。
    在实际应用中，这里会是调用Google Maps API等。
    """
    time.sleep(0.01) # 模拟API请求的延迟，例如10毫秒
    # 模拟返回经纬度数据
    lat = np.random.uniform(30, 40)
    lon = np.random.uniform(-100, -90)
    return f"Lat:{lat:.4f}, Lon:{lon:.4f}"

# 定义批次数据处理函数
def process_data_chunk(chunk_df):
    """
    对单个数据批次执行复杂的apply操作和API调用。
    """
    # 示例1: 执行一个复杂的apply操作
    chunk_df['processed_value'] = chunk_df['value1'] * 0.5 + chunk_df['value2'] / 10

    # 示例2: 对地址列进行API调用
    # 注意：如果API支持批量查询，应优先使用批量查询以减少网络开销
    # 这里为了演示，我们假设API是按行调用的
    chunk_df['coordinates'] = chunk_df['address'].apply(get_coordinates_from_address)

    # 示例3: 模拟一个merge操作 (如果需要与其他DataFrame合并)
    # 假设有一个小型配置DataFrame需要合并
    # config_df = pd.DataFrame({'id': [0, 1, 2], 'config_info': ['A', 'B', 'C']})
    # chunk_df = pd.merge(chunk_df, config_df, on='id', how='left')

    return chunk_df

output_csv_path = 'processed_large_dataframe.csv'

# 如果输出文件已存在，先删除，确保从头开始写入
if os.path.exists(output_csv_path):
    os.remove(output_csv_path)
    print(f"已删除旧的输出文件: {output_csv_path}")

header_written = False # 标记是否已写入CSV头部

print(f"\n开始分批处理 {len(df)} 行数据并写入 {output_csv_path}...")

unique_batches = df['batch_num'].unique()
total_batches = len(unique_batches)

for i, batch_id in enumerate(unique_batches):
    # 提取当前批次的数据
    current_batch_df = df[df['batch_num'] == batch_id].copy() # 使用 .copy() 避免 SettingWithCopyWarning

    print(f"正在处理批次 {i+1}/{total_batches} (行范围: {current_batch_df.index.min()} - {current_batch_df.index.max()})")

    # 处理当前批次的数据
    processed_batch = process_data_chunk(current_batch_df)

    # 将处理后的批次数据写入CSV文件
    if not header_written:
        # 首次写入，包含头部
        processed_batch.to_csv(output_csv_path, mode='w', index=False, encoding='utf-8')
        header_written = True
    else:
        # 后续写入，不包含头部，以追加模式写入
        processed_batch.to_csv(output_csv_path, mode='a', header=False, index=False, encoding='utf-8')

    # 可选：在批次之间引入额外的延迟，以更严格地遵守API速率限制
    # time.sleep(0.5) # 例如，每处理完一个批次暂停0.5秒

print(f"\n所有批次处理完成，结果已写入 {output_csv_path}")

# 验证写入结果 (可选)
# processed_df = pd.read_csv(output_csv_path)
# print(f"\n从CSV读取的数据总行数: {len(processed_df)}")
# print("前5行数据示例:")
# print(processed_df.head())