处理未完整关闭的Parquet文件：解决Footer魔术字节缺失问题

一、读取不完整Parquet文件（仅加载完整Row Group） #

默认的pd.read_parquet会严格校验文件末尾的Parquet magic bytes和footer，遇到未完整关闭的文件直接报错。要仅加载已完成的Row Group，需要用pyarrow的低级API手动解析：

import pyarrow as pa
import pyarrow.parquet as pq
import pandas as pd

def read_valid_row_groups(parquet_path):
    with pa.memory_map(parquet_path, 'r') as source:
        try:
            # 优先尝试读取完整文件
            return pq.read_table(source).to_pandas()
        except pa.ArrowInvalid as e:
            if "magic bytes not found in footer" not in str(e):
                raise  # 非footer缺失错误，直接抛出
            
            # 手动遍历并读取完整的Row Group
            parquet_file = pq.ParquetFile(source, validate_schema=False)
            valid_groups = []
            
            for idx in range(parquet_file.num_row_groups):
                try:
                    rg_table = parquet_file.read_row_group(idx)
                    valid_groups.append(rg_table)
                except pa.ArrowInvalid:
                    print(f"警告：跳过不完整的Row Group {idx}")
                    continue
            
            if not valid_groups:
                raise ValueError("文件中无完整的Row Group可读取")
            
            combined_table = pa.concat_tables(valid_groups)
            print("警告：文件未完整关闭，仅加载已写入的完整Row Group")
            return combined_table.to_pandas()

使用时直接调用read_valid_row_groups('path/to/your/file.parquet')即可，会返回包含所有完整Row Group的DataFrame，并输出缺失footer的警告。

二、C++端写入时确保Row Group即时可读取 #

要让Parquet文件在写入过程中（或崩溃后）仍能被读取到完整的Row Group，核心是确保每个Row Group的数据和元数据即时刷入磁盘，而不是留在内存缓存中。

关键实现步骤： #

1. 配置写入器属性，开启Row Group后自动刷新
2. 写完每个Row Group后手动触发磁盘同步
3. 最终正常关闭时写入完整footer

C++代码示例：

#include <arrow/record_batch.h>
#include <parquet/arrow/writer.h>

int main() {
    // 打开输出文件
    std::shared_ptr<arrow::io::FileOutputStream> out_stream;
    PARQUET_THROW_NOT_OK(arrow::io::FileOutputStream::Open("target.parquet", &out_stream));

    // 配置写入器属性：Row Group写入后自动刷新，禁用异步写入
    parquet::WriterProperties::Builder prop_builder;
    prop_builder.flush_after_row_group(true);
    prop_builder.enable_async_write(false);
    auto writer_props = prop_builder.build();

    // 创建Parquet写入器（your_schema为你的数据结构）
    std::shared_ptr<parquet::arrow::FileWriter> parquet_writer;
    PARQUET_THROW_NOT_OK(
        parquet::arrow::FileWriter::Open(your_schema, arrow::default_memory_pool(), out_stream, writer_props, &parquet_writer)
    );

    // 循环写入每个RecordBatch（每个Batch对应一个Row Group）
    for (const auto& batch : your_record_batches) {
        // 写入当前Batch
        PARQUET_THROW_NOT_OK(parquet_writer->WriteRecordBatch(batch));
        // 手动刷新到磁盘，确保数据不留在缓存
        PARQUET_THROW_NOT_OK(parquet_writer->Flush());
        PARQUET_THROW_NOT_OK(out_stream->Sync());
    }

    // 正常关闭时写入完整footer
    PARQUET_THROW_NOT_OK(parquet_writer->Close());
    return 0;
}

原理说明： #

Parquet的Row Group元数据会紧跟在对应的数据块之后（或存储在文件的元数据区域），即使没有最终的footer，只要Row Group的数据和元数据已刷入磁盘，读取端就可以通过扫描文件找到这些完整的Row Group。上述写入逻辑确保每个Row Group完成后立即持久化，避免崩溃时丢失或损坏Row Group数据。

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