如何在无本地磁盘的情况下将S3/MinIO中50GB+压缩CSV流转为Parquet格式

核心思路 #

要避免依赖计算单元的本地存储，核心是全程流式处理——不下载完整压缩包、不解压到本地、不落地中间文件，所有操作都在内存流中完成，逐块处理数据。

具体实现步骤 #

• 流式对接对象存储：使用支持流式读写的SDK（如boto3、minio），直接从源存储桶获取压缩包的字节流，同时将转换后的Parquet字节流直接上传到MinIO。
• 流式解压归档：利用Python标准库的zipfile/tarfile直接操作字节流，无需将整个归档加载到内存或本地。
• 分块处理CSV转Parquet：对每个CSV文件采用分块读取+流式写入Parquet的方式，控制内存占用，适配大文件场景。

关键代码示例 #

1. 初始化存储客户端

import boto3
from minio import Minio
from io import BytesIO

# 源存储桶客户端（兼容S3协议的存储均可）
source_s3 = boto3.client(
    's3',
    endpoint_url='<源存储端点>',
    aws_access_key_id='<源访问密钥>',
    aws_secret_access_key='<源密钥>'
)

# MinIO客户端
minio_client = Minio(
    '<MinIO端点>',
    access_key='<MinIO访问密钥>',
    secret_key='<MinIO密钥>',
    secure=False  # 根据实际部署配置调整
)

2. 流式读取并解压归档包

以ZIP归档为例，直接从源存储读取字节流，逐个处理内部CSV文件：

import zipfile

# 从源存储获取压缩包的流式响应
response = source_s3.get_object(Bucket='<源存储桶名>', Key='<压缩包路径>')
zip_stream = response['Body']

# 流式打开ZIP包，不加载整个归档到内存
with zipfile.ZipFile(zip_stream, 'r') as zf:
    # 遍历归档内的所有文件
    for file_name in zf.namelist():
        # 只处理CSV文件
        if not file_name.endswith('.csv'):
            continue
        
        # 获取CSV文件的流式对象
        with zf.open(file_name) as csv_stream:
            # 处理CSV转Parquet并上传到MinIO
            process_csv_to_parquet(csv_stream, file_name, minio_client)

3. 分块处理CSV转Parquet并流式上传

使用pandas分块读取CSV，结合pyarrow流式写入Parquet，直接将数据流上传到MinIO：

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

def process_csv_to_parquet(csv_stream, original_name, minio_client):
    # 生成Parquet文件名（替换后缀）
    parquet_name = original_name.replace('.csv', '.parquet')
    # 创建内存字节流用于存储Parquet数据
    parquet_stream = BytesIO()
    
    # 分块读取CSV，控制内存占用
    chunk_size = 10000  # 可根据计算节点内存调整
    for chunk in pd.read_csv(csv_stream, chunksize=chunk_size):
        # 可选：添加业务相关的数据清洗/转换逻辑
        # chunk = chunk.dropna(subset=['key_column'])
        
        # 将DataFrame转为Arrow表
        table = pa.Table.from_pandas(chunk)
        
        # 流式写入Parquet（第一次写初始化文件，后续追加）
        if parquet_stream.tell() == 0:
            pq.write_table(table, parquet_stream)
        else:
            parquet_stream.seek(0, 2)
            pq.write_table(table, parquet_stream, append=True)
    
    # 将流指针重置到开头，准备上传
    parquet_stream.seek(0)
    
    # 上传Parquet流到MinIO
    minio_client.put_object(
        Bucket='<目标MinIO桶名>',
        ObjectName=f'parquet/{parquet_name}',
        data=parquet_stream,
        length=parquet_stream.getbuffer().nbytes,
        content_type='application/octet-stream'
    )

优化建议 #

• 归档格式适配：如果是Tar.gz归档，用tarfile.open(mode='r:gz', fileobj=tar_stream)实现流式解压，逻辑与ZIP一致。
• 内存优化：对于超大型CSV，可替换为pyarrow.csv的流式读取API，比pandas更节省内存；根据计算节点内存灵活调整chunk_size。
• 并行处理：用concurrent.futures.ThreadPoolExecutor并行处理多个CSV文件，每个文件的流独立处理即可保证线程安全。
• 错误处理：添加try-except块捕获网络异常、文件格式错误，记录失败的文件名和错误信息，支持后续重试。
• Spark兼容性：Parquet写入时指定version='2.6'（Spark默认兼容版本），确保Schema与Spark读取逻辑匹配。

内容的提问来源于stack exchange，提问作者Роман Кирьянов