如何实现Pandas DataFrame分块写入String Buffer并上传至S3？

你的问题出在对pandas to_csv方法中chunksize参数的理解上——这个参数在写入到StringIO这类内存对象时不会自动把DataFrame拆分成多个块，它原本是用于写入磁盘文件时控制单次写入的行数缓冲区，而非分割DataFrame。所以你的代码只会把整个DataFrame（或你误以为的前1000行）一次性写入缓冲区，无法实现分块上传的需求。

下面给出两种符合你需求的解决方案：

方案一：将CSV拆分为多个小文件上传到S3 #

如果你的需求是把10000行数据拆成10个1000行的独立CSV文件上传，这个方案最直接：

import pandas as pd
import boto3
from io import StringIO

# 假设你已经有了10000行的DataFrame df
# df = pd.read_csv("your_source_file.csv")

s3_resource = boto3.resource('s3')
bucket_name = "你的S3桶名称"
chunk_size = 1000

# 把大DataFrame拆分成多个子DataFrame
df_chunks = [df[i:i+chunk_size] for i in range(0, len(df), chunk_size)]

for chunk_idx, chunk_df in enumerate(df_chunks):
    csv_buffer = StringIO()
    # 只有第一个分块写入表头，后续分块跳过表头避免重复
    chunk_df.to_csv(csv_buffer, index=False, header=(chunk_idx == 0))
    # 上传到S3，文件命名为 df_part_0.csv、df_part_1.csv...
    s3_object_key = f"df_part_{chunk_idx}.csv"
    s3_resource.Object(bucket_name, s3_object_key).put(Body=csv_buffer.getvalue())
    print(f"分块 {chunk_idx+1}/{len(df_chunks)} 上传完成")

方案二：用S3分片上传（Multipart Upload）合并为单个文件 #

如果需要把所有数据合并成一个CSV文件，但通过分块的方式上传（适合超大文件，避免内存溢出），可以用S3的Multipart Upload功能：

import pandas as pd
import boto3
from io import StringIO

# 假设你已经有了10000行的DataFrame df
# df = pd.read_csv("your_source_file.csv")

s3_client = boto3.client('s3')
bucket_name = "你的S3桶名称"
target_file_key = "df.csv"
chunk_size = 1000

# 初始化分片上传
multipart_response = s3_client.create_multipart_upload(Bucket=bucket_name, Key=target_file_key)
upload_id = multipart_response['UploadId']

uploaded_parts = []
df_chunks = [df[i:i+chunk_size] for i in range(0, len(df), chunk_size)]

for chunk_idx, chunk_df in enumerate(df_chunks):
    csv_buffer = StringIO()
    # 控制表头：仅第一个分块写入表头
    chunk_df.to_csv(csv_buffer, index=False, header=(chunk_idx == 0))
    # 转成字节流（S3上传需要bytes类型）
    chunk_content = csv_buffer.getvalue().encode('utf-8')
    
    # 上传当前分片
    part_response = s3_client.upload_part(
        Bucket=bucket_name,
        Key=target_file_key,
        PartNumber=chunk_idx + 1,
        UploadId=upload_id,
        Body=chunk_content
    )
    uploaded_parts.append({
        'PartNumber': chunk_idx + 1,
        'ETag': part_response['ETag']
    })
    print(f"分片 {chunk_idx+1}/{len(df_chunks)} 上传完成")

# 完成分片上传，合并所有分片为单个文件
s3_client.complete_multipart_upload(
    Bucket=bucket_name,
    Key=target_file_key,
    UploadId=upload_id,
    MultipartUpload={'Parts': uploaded_parts}
)

print("全部分片上传完成，文件已合并！")

关键说明 #

• 手动拆分DataFrame：用列表推导式[df[i:i+chunk_size]...]是最直接的方式，能精准控制每个分块的行数。
• 表头处理：必须确保只有第一个分块写入表头，否则最终的CSV会出现重复表头，导致数据解析错误。
• Multipart Upload的优势：对于GB级以上的超大文件，这种方式可以避免一次性加载所有数据到内存，同时S3会自动处理分片的合并，最终得到完整的单个文件。

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