使用Python无需全量加载进内存即可读写Parquet文件的实现方法

方案1：PyArrow分块读取写入（支持按需筛选列） #

该方案完全基于Python SDK实现，支持仅读取指定列，全程不会加载全量文件到内存。

依赖安装 #

pip install pyarrow fsspec adlfs

实现代码 #

import pyarrow.parquet as pq
from fsspec.filesystem import HTTPFileSystem
from adlfs import AzureBlobFileSystem

# 配置参数
SOURCE_URL = "https://my-file-location.parquet"
TARGET_CONTAINER = "你的Azure容器名"
TARGET_PATH = "my-file.parquet"
# Azure存储认证，根据实际情况替换
AZURE_ACCOUNT_NAME = "你的存储账户名"
AZURE_ACCOUNT_KEY = "你的存储账户密钥"
# 按需配置：需要读取的列，填None则读取全部列
USE_COLS = ["col1", "col2", "col3"]
# 按需配置：单次读取的行数，可根据内存大小调整
BATCH_SIZE = 10000

# 初始化源端HTTP文件系统
fs_http = HTTPFileSystem()
# 初始化目标端Azure存储文件系统
fs_azure = AzureBlobFileSystem(
    account_name=AZURE_ACCOUNT_NAME,
    account_key=AZURE_ACCOUNT_KEY
)

# 打开源Parquet文件
with fs_http.open(SOURCE_URL, mode="rb") as src_f:
    parquet_file = pq.ParquetFile(src_f)
    # 打开目标端写入流
    with fs_azure.open(f"{TARGET_CONTAINER}/{TARGET_PATH}", mode="wb") as tgt_f:
        # 初始化Parquet写入器，schema和源文件一致
        with pq.ParquetWriter(tgt_f, schema=parquet_file.schema.to_arrow_schema()) as writer:
            # 按批次迭代读取，支持指定列和批次大小
            for batch in parquet_file.iter_batches(columns=USE_COLS, batch_size=BATCH_SIZE):
                writer.write_batch(batch)

说明 #

• Parquet的iter_batches接口会仅拉取你指定的列数据，不需要下载全文件，100列里只取3列的场景下可以节省97%左右的流量和读取耗时
• 全程只有当前批次的数据驻留内存，不会加载全量文件，可通过调整BATCH_SIZE适配不同的内存限制

方案2：整文件流式拷贝（无需解析Parquet） #

如果你不需要筛选列，只是要做文件迁移，不需要解析Parquet结构，可以直接用fsspec的文件流拷贝，性能更高：

from fsspec.filesystem import HTTPFileSystem
from adlfs import AzureBlobFileSystem

SOURCE_URL = "https://my-file-location.parquet"
TARGET_CONTAINER = "你的Azure容器名"
TARGET_PATH = "my-file.parquet"
AZURE_ACCOUNT_NAME = "你的存储账户名"
AZURE_ACCOUNT_KEY = "你的存储账户密钥"
# 单次拷贝的块大小
CHUNK_SIZE = 10*1024*1024 # 10MB

fs_http = HTTPFileSystem()
fs_azure = AzureBlobFileSystem(
    account_name=AZURE_ACCOUNT_NAME,
    account_key=AZURE_ACCOUNT_KEY
)

with fs_http.open(SOURCE_URL, "rb") as src_f, fs_azure.open(f"{TARGET_CONTAINER}/{TARGET_PATH}", "wb") as tgt_f:
    while chunk := src_f.read(CHUNK_SIZE):
        tgt_f.write(chunk)

该方案完全跳过Parquet解析，直接流式传输字节流，内存占用稳定在CHUNK_SIZE设定的大小。

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