import pandas as pd

# 分块读取大型CSV文件
chunk_size = 10000  # 每块包含10,000行
chunks = pd.read_csv('large_dataset.csv', chunksize=chunk_size)

for i, chunk in enumerate(chunks):
    # 处理每个数据块
    processed_chunk = process_data(chunk)
    
    # 如果是第一次迭代，创建新文件，否则追加
    if i == 0:
        processed_chunk.to_csv('processed_data.csv', index=False)
    else:
        processed_chunk.to_csv('processed_data.csv', mode='a', header=False, index=False)
    
    print(f"Processed chunk {i+1}")
                

优势：

• 避免一次性加载整个文件导致内存溢出
• 可以实时查看处理进度
• 处理失败时可以从中断点恢复

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 创建数据库连接
engine = create_engine('postgresql://user:password@localhost/db')
Session = sessionmaker(bind=engine)
session = Session()

# 分块大小
chunk_size = 5000
offset = 0

while True:
    # 查询一个数据块
    results = session.query(User).order_by(User.id).offset(offset).limit(chunk_size).all()
    
    if not results:
        break  # 没有更多数据
    
    # 处理数据
    process_users(results)
    
    # 更新偏移量
    offset += chunk_size
    print(f"Processed {offset} records")
    
session.close()
                

import requests
import time

def chunked_api_call(data, chunk_size=50, delay=1):
    """分块调用API，避免速率限制"""
    results = []
    
    for i in range(0, len(data), chunk_size):
        chunk = data[i:i + chunk_size]
        
        # 准备API请求
        payload = {"items": chunk}
        response = requests.post("https://api.example.com/process", json=payload)
        
        if response.status_code == 200:
            results.extend(response.json()['results'])
            print(f"Processed {len(chunk)} items, total: {i+len(chunk)}")
        else:
            print(f"Error on chunk {i//chunk_size}: {response.status_code}")
        
        # 延迟以避免触发速率限制
        time.sleep(delay)
    
    return results

# 使用示例
large_data = [...]  # 假设有10,000条数据
results = chunked_api_call(large_data, chunk_size=100, delay=0.5)
                

错误处理技巧：

• 添加重试机制处理临时错误
• 记录失败的数据块以便后续重试
• 使用指数退避策略处理速率限制
• 设置超时时间防止请求挂起