无需读取全部数据向AWS S3的parquet文件追加行的Python实现方法

可以在不加载全量原始Parquet文件到内存的前提下完成追加操作，但要注意Parquet是列式二进制存储格式，本身不支持原地追加内容，你需要生成合并后的新Parquet文件（可直接覆盖原文件），整个过程按行组流式读写，内存占用仅和单条行组大小有关，和原文件总大小无关，完全适配AWS Lambda的运行限制。

推荐使用pyarrow + s3fs实现，无需将原文件下载到本地磁盘，全程基于S3流式读写，符合Lambda的资源限制。你可以提前将两个库打包为Lambda层，也可以直接使用AWS官方预置了数据处理依赖的运行时。

代码示例 #

import pyarrow.parquet as pq
import pyarrow as pa
import s3fs

# 初始化S3文件系统，会自动读取Lambda运行时的IAM权限
s3 = s3fs.S3FileSystem()

def append_parquet_s3(source_path: str, new_data: list[dict], target_path: str = None):
    # 如果不传target_path则直接覆盖原文件
    target_path = target_path or source_path
    
    # 仅读取原文件的schema，无需加载全量数据
    source_schema = pq.read_schema(source_path, filesystem=s3)
    
    # 将待追加数据转为符合schema的表结构
    new_table = pa.Table.from_pylist(new_data, schema=source_schema)
    
    # 创建流式写入器
    with pq.ParquetWriter(
        where=target_path,
        schema=source_schema,
        filesystem=s3,
        compression='snappy' # 保持和原文件一致的压缩格式即可
    ) as writer:
        # 按行组迭代读取原文件，每读一个行组直接写入目标文件，无全量缓存
        source_parquet = pq.ParquetFile(source_path, filesystem=s3)
        for rg_idx in range(source_parquet.num_row_groups):
            row_group = source_parquet.read_row_group(rg_idx)
            writer.write_table(row_group)
        
        # 写入新增数据
        writer.write_table(new_table)

# 调用示例
if __name__ == "__main__":
    # 待追加的数据，结构必须和原Parquet schema完全一致
    append_rows = [
        {"id": 1001, "name": "test1", "value": 123},
        {"id": 1002, "name": "test2", "value": 456}
    ]
    append_parquet_s3(
        source_path="s3://你的桶名/路径/原文件.parquet",
        new_data=append_rows
    )

优化建议 #

• 如果你没有强制要求合并为单个Parquet文件，更推荐将新增数据写入为独立的小Parquet文件，和原文件放在相同的S3前缀下。后续用Athena、Spark、Pandas等工具查询该前缀时，会自动识别所有同schema的Parquet文件，无需任何合并操作，这种方案资源消耗最低，完全避免重写原文件的开销，也不受Lambda运行时长限制。
• 如果必须合并为单文件，确保Lambda的内存配置不小于原Parquet单条行组的大小，常规Parquet行组大小为128M~512M，Lambda默认配置即可支持；如果原文件大小超过10G，不推荐在Lambda中做单文件合并，避免超过Lambda最长15分钟的运行时长限制。

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