如何为Python Requests添加并行处理以加速10万+数据拉取

首先修正你原代码中的明显错误：请求URL里的str(r.status_code)完全不合理，因为r还未定义就调用其属性，这会直接触发报错，已在下方代码中修复。

针对10万条数据的分批并行拉取需求，这里用线程池实现（IO密集型任务用线程池比进程池更高效），每批拉取1000条，具体代码如下：

#!/usr/bin/env python
import requests
from concurrent.futures import ThreadPoolExecutor

# 基础配置
API_URL = "your_server_url_here"  # 替换为实际服务器地址
TOKEN = "server_data_token"
BATCH_SIZE = 1000  # 每批拉取1000条
MAX_WORKERS = 5  # 并行线程数，根据服务器限流规则调整，别设置过高

def fetch_batch(offset):
    """拉取指定偏移量的一批数据"""
    data = {
        'token': TOKEN,
        'content': 'record',
        'format': 'csv',
        'type': 'flat',
        'csvDelimiter': '',
        'rawOrLabel': 'raw',
        'rawOrLabelHeaders': 'raw',
        'exportCheckboxLabel': 'false',
        'exportSurveyFields': 'false',
        'exportDataAccessGroups': 'false',
        'returnFormat': 'json',
        'offset': offset,
        'limit': BATCH_SIZE
    }
    try:
        response = requests.post(API_URL, data=data)
        response.raise_for_status()  # 捕获HTTP请求错误
        return response.text
    except requests.exceptions.RequestException as e:
        print(f"拉取偏移量{offset}时出错: {e}")
        return None

def main():
    # 先获取总记录数，计算需要分多少批
    count_data = {
        'token': TOKEN,
        'content': 'record',
        'format': 'count',  # 假设API支持该参数返回总记录数
        'returnFormat': 'json'
    }
    total_records = requests.post(API_URL, data=count_data).json()
    print(f"总记录数: {total_records}")

    # 生成所有需要拉取的偏移量
    offsets = list(range(0, total_records, BATCH_SIZE))

    # 并行拉取所有批次
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        results = list(executor.map(fetch_batch, offsets))

    # 合并CSV结果，跳过重复表头
    merged_csv = ""
    header_written = False
    for batch in results:
        if not batch:
            continue
        lines = batch.splitlines()
        if not header_written:
            merged_csv += "\n".join(lines) + "\n"
            header_written = True
        else:
            # 跳过表头，只追加数据行
            merged_csv += "\n".join(lines[1:]) + "\n"

    # 保存合并后的数据到文件
    with open("full_data.csv", "w", encoding="utf-8") as f:
        f.write(merged_csv)
    print("数据拉取完成，已保存到full_data.csv")

if __name__ == "__main__":
    main()

关键说明 #

• 分批逻辑：通过offset（起始偏移量）和limit（每批条数）参数实现分页拉取，这是绝大多数数据导出API都支持的参数，如果你的API参数名不同，对应修改即可。
• 并行处理：用ThreadPoolExecutor启动多线程并发请求，避免单线程等待响应的时间浪费，MAX_WORKERS建议从5开始测试，过高可能触发服务器限流。
• 异常处理：单个批次请求失败不会中断整个任务，只会打印错误信息。
• CSV合并：自动跳过重复的表头，保证最终生成的CSV格式正确。

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