如何用boto3拆分Parquet文件为最多1000行并上传至S3

要实现将Parquet文件拆分为最多1000行的小文件再上传到S3，你需要先完成Parquet文件的行级拆分，再结合现有的save_to_s3函数完成上传。以下是完整的实现方案：

1. 依赖准备 #

首先确保安装处理Parquet文件的依赖库（推荐用pyarrow，性能更优）：

pip install pyarrow

2. 拆分Parquet并上传的完整代码 #

import pyarrow.parquet as pq
from boto3.session import Session
import config  # 你的配置模块

def split_parquet_and_upload(local_parquet_path: str, bucket: str, s3_base_path: str, max_rows_per_file: int = 1000):
    # 读取原Parquet文件
    parquet_file = pq.ParquetFile(local_parquet_path)
    
    # 按指定行数逐批次拆分
    for batch_idx, batch in enumerate(parquet_file.iter_batches(batch_size=max_rows_per_file)):
        # 将批次转换为Parquet字节流
        table = batch.to_table()
        buffer = table.write_to_buffer()
        parquet_bytes = buffer.getvalue()
        
        # 生成S3存储路径（沿用你原有的序号命名规则）
        file_path = f"{s3_base_path}{batch_idx:06}.parquet"
        # 调用上传函数
        save_to_s3(bucket, file_path, parquet_bytes, content_type="application/octet-stream")

def save_to_s3(bucket: str | object,
               path: str,
               data: bytes | str,
               tags: dict[str, str] = None,
               content_type: str = "",
               metadata: dict[str, str] = None) -> None:
    # 优化Bucket实例初始化：若传入字符串则自动创建
    if isinstance(bucket, str):
        bucket = Session(region_name=config.aws_region).resource('s3').Bucket(bucket)

    # 合并元数据并添加创建者标识
    metadata = {**(metadata or {}), **{'created_by': config.spark_job_name}}
    data_hash = _get_content_hash(data)
    tags = _encode_object_tags(tags or {})

    bucket.put_object(
        Key=path,
        Body=data,
        ContentMD5=data_hash,
        Metadata=metadata,
        Tagging=tags,
        ContentType=content_type,
    )

# 辅助函数（假设你已实现以下逻辑，若未实现可参考示例代码）
def _get_content_hash(data: bytes | str) -> str:
    import hashlib
    import base64
    if isinstance(data, str):
        data = data.encode('utf-8')
    md5 = hashlib.md5(data).digest()
    return base64.b64encode(md5).decode('utf-8')

def _encode_object_tags(tags: dict[str, str]) -> str:
    return '&'.join([f"{k}={v}" for k, v in tags.items()])

关键说明 #

• 拆分逻辑：借助pyarrow的iter_batches方法，按1000行的粒度逐块读取原Parquet文件，每块生成独立的Parquet字节流，避免一次性加载大文件到内存。
• 文件命名：沿用你原有的6位数字序号命名规则，保证拆分后的文件有序且唯一。
• 上传适配：将拆分后的Parquet字节流直接传入save_to_s3，指定content_type为application/octet-stream（Parquet文件标准MIME类型）。
• 函数优化：调整save_to_s3的Bucket参数处理逻辑，支持传入字符串或已实例化的Bucket对象，避免重复初始化Session。

使用示例 #

# 本地大Parquet文件路径
local_file = "/data/large_dataset.parquet"
# S3存储桶名称
s3_bucket = "your-target-bucket"
# S3文件前缀路径
s3_prefix = "processed/parquet_split/"

# 执行拆分上传
split_parquet_and_upload(local_file, s3_bucket, s3_prefix)

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