为何读取Parquet数据集所需内存远超其大小？PyArrow单节点内存溢出问题

我用Python的PyArrow库从S3读取Parquet数据集，S3显示该路径下数据集大小为14.3GB，共包含836个对象。我在拥有64GB内存的c4.8xlarge EC2实例上运行代码，尽管内存是数据集大小的4倍以上，但程序仍因内存耗尽崩溃。

我想知道：为什么读取该数据集需要如此多的内存？有没有办法在单节点工作流中避免这个问题？我了解Spark、Dask等分布式计算库，且能在PySpark中正常使用该数据集，但希望搭建单节点工作流。

读取数据集的代码如下：

import pyarrow.parquet as pq
from pyarrow import fs
s3 = fs.S3FileSystem()

#fs = s3fs.S3FileSystem()
bucket = "<bucket_name>"
path = "<path>"

dataset = pq.ParquetDataset(f"{bucket}/{path}", filesystem=s3)

数据集Schema及统计信息摘要（我仅读取113列中的9列，共7,045,204行）：

• Column 1: int
• Column 2: Array，平均长度约450
• Column 3: Array，平均长度约450
• Column 4: Array，平均长度约1000
• Column 5: Array，平均长度约1000
• Column 6: String，平均长度2
• Column 7: int
• Column 8: int
• Column 9: timestamp

1. Parquet压缩特性：S3显示的14.3GB是压缩后的磁盘大小，PyArrow读取时需要将数据全量解压到内存。你的数据中包含大量数组列，这类数据的压缩比通常很高，解压后的实际内存占用会远大于原始压缩体积。
2. 数组列的内存开销：PyArrow中数组类型以连续内存块存储，每个int类型元素至少占用4字节。按你的统计计算：
   • Column2+Column3：每行共约900个int，700万行对应约63亿个元素，仅这两列就需要约25.2GB内存
   • Column4+Column5：每行共约2000个int，700万行对应约140亿个元素，需要约56GB内存
     加上其他列的开销，总内存需求轻松超过64GB，直接导致内存耗尽。
3. 默认读取行为：pq.ParquetDataset默认会尝试将全量数据加载到内存（如果后续执行dataset.read()），即使只指定部分列，数组列的解压和存储依然会占用大量内存。

1. 确认按需列读取生效 #

确保读取时仅加载需要的9列，避免冗余列占用内存，可在read()方法中明确指定列：

table = dataset.read(columns=["column1", "column2", ...])  # 替换为实际列名

2. 分块读取并处理数据 #

利用PyArrow的分片能力，将数据集拆分为小片段逐块处理，避免一次性加载全量数据：

# 获取数据集分片
fragments = dataset.fragments

for fragment in fragments:
    # 逐片读取目标列
    table = fragment.read(columns=["column1", "column2", ...])
    # 处理当前分片数据（如转换为pandas DataFrame后计算）
    df = table.to_pandas()
    # 处理完成后，table和df会被垃圾回收，释放内存

3. 限制PyArrow内存池大小 #

通过设置PyArrow内存池的最大限制，避免内存无上限占用：

import pyarrow as pa
# 设置内存池最大限制为32GB
pa.set_memory_pool_limit(32 * 1024**3)

4. 转换为pandas时用高效数据类型 #

如果需要转换为pandas DataFrame，可对数组列使用更节省内存的类型（如数据范围允许，将int转为int32）：

df = table.to_pandas()
# 转换数组列的数值类型
df["column2"] = df["column2"].astype("int32")

5. 用Scanner API批量处理数据 #

利用ParquetDataset的扫描能力，按批次获取数据并增量处理，无需全量加载：

scanner = dataset.scanner(columns=["column1", "column2", ...])
# 按批次读取数据，批次大小可根据内存调整
for batch in scanner.to_batches(batch_size=100000):
    # 处理当前批次的数据
    process_batch(batch)

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