从S3按日期/大小过滤加载Parquet文件遇ArrowInvalid错误求助

尝试从S3加载Parquet文件时，通过Range参数仅读取前1MB数据，同时按日期过滤，代码如下：

obj = s3_client.get_object(Bucket=bucket_name, Key=file_path1, 
Range=f'bytes=0-1048576')

filters = [('date', '>=', start_date), ('date', '<=', end_date)]

body = io.BytesIO(obj['Body'].read())

df = pd.read_parquet((body), engine='pyarrow', filters=filters)

执行时触发错误：

ArrowInvalid: Could not open Parquet input source '': Parquet magic bytes not found in footer.

Parquet文件的元数据（包括footer、索引等）存储在文件末尾，你用Range只读取了文件开头部分，缺失了Parquet解析必需的footer魔法字节和元数据，导致pyarrow无法识别这是有效的Parquet文件，自然无法解析和应用过滤条件。同时，按日期过滤需要依赖Parquet的分区信息或文件内的元数据索引，仅读取部分字节也无法获取这些过滤所需的核心数据。

方案1：读取完整文件后过滤（适合小文件） #

如果文件体积不大，直接读取完整文件再应用过滤逻辑：

obj = s3_client.get_object(Bucket=bucket_name, Key=file_path1)
body = io.BytesIO(obj['Body'].read())
df = pd.read_parquet(body, engine='pyarrow', filters=filters)

方案2：利用S3 Select在服务端过滤（推荐大文件） #

使用S3 Select功能让AWS在服务端完成数据过滤，仅返回符合条件的数据，大幅减少传输量：

response = s3_client.select_object_content(
    Bucket=bucket_name,
    Key=file_path1,
    ExpressionType='SQL',
    Expression=f"SELECT * FROM s3object s WHERE s.date >= '{start_date}' AND s.date <= '{end_date}'",
    InputSerialization={'Parquet': {}},
    OutputSerialization={'CSV': {}}  # 也可选择JSON格式，后续转换为DataFrame
)

# 处理返回的数据流
records = []
for event in response['Payload']:
    if 'Records' in event:
        records.append(event['Records']['Payload'])

# 拼接数据并转为DataFrame
df = pd.read_csv(io.BytesIO(b''.join(records)))

注意：需确保Parquet文件的date字段可被S3 Select识别，若字段为日期格式，可能需要调整SQL表达式中的日期写法。

方案3：按Parquet数据块读取（进阶） #

Parquet文件由多个数据块（row group）组成，可先读取文件末尾的元数据，判断哪些数据块包含目标日期范围，再仅读取这些块：

import pyarrow.parquet as pq
import pyarrow.fs as pfs
from io import BytesIO

# 使用pyarrow的S3文件系统直接操作
fs = pfs.S3FileSystem()
with fs.open(file_path1, 'rb') as f:
    parquet_file = pq.ParquetFile(f)
    filtered_row_groups = []
    # 遍历每个row group，通过统计信息判断是否包含目标日期
    for i, rg in enumerate(parquet_file.row_groups):
        # 假设date是第一个字段，且文件写入时开启了统计功能
        min_date = rg.column(0).statistics.min
        max_date = rg.column(0).statistics.max
        if max_date >= start_date and min_date <= end_date:
            filtered_row_groups.append(i)
    # 仅读取符合条件的row groups
    df = parquet_file.read_row_groups(filtered_row_groups, filters=filters).to_pandas()

此方案要求Parquet文件在写入时开启了统计信息功能，否则无法提前判断数据块是否包含目标数据。

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