百万小DataFrame并行存Parquet：能否共用单个元信息文件？

针对你遇到的小DataFrame批量存储元数据重复浪费空间的问题，以下是几种可落地的实现方案，兼顾存储效率与并行处理需求：

方案1：复用Parquet模板元数据（推荐） #

Parquet原生支持元数据复用，你可以先生成一个包含完整Schema的模板文件，后续所有小DataFrame写入时直接复用该模板的元数据，仅写入数据部分。

操作步骤： #

1. 生成Schema模板文件

   import pandas as pd
   # 替换为你实际的DataFrame结构（列名、数据类型、索引）
   template_df = pd.DataFrame(columns=['col1', 'col2', 'col3'], dtype='float64')
   # 保存包含完整元数据的模板文件
   template_df.to_parquet('metadata_template.parquet', engine='pyarrow')
   

2. 并行写入时复用元数据
   使用pyarrow底层API跳过重复元数据写入：

   import pyarrow as pa
   import pyarrow.parquet as pq
   
   # 预加载模板元数据
   template_metadata = pq.read_metadata('metadata_template.parquet')
   
   def save_small_df(df, file_path):
       table = pa.Table.from_pandas(df)
       # 复用模板元数据，关闭统计信息写入进一步缩减体积
       pq.write_table(table, file_path, metadata=template_metadata, write_statistics=False)
   

方案2：自定义数据+独立元文件 #

如果需要更强的灵活性，可以拆分元数据与数据存储，自己维护结构一致性：

操作步骤： #

1. 单独保存元数据

   # 从模板DataFrame导出结构信息
   metadata = {
       'columns': list(template_df.columns),
       'dtypes': template_df.dtypes.astype(str).to_dict(),
       'index_names': list(template_df.index.names) if template_df.index.names[0] else []
   }
   import json
   with open('df_metadata.json', 'w') as f:
       json.dump(metadata, f)
   

2. 仅存储原始数据

   def save_small_df_data(df, file_path):
       table = pa.Table.from_pandas(df)
       # 用pyarrow IPC格式存储纯数据，比CSV更高效
       with pa.OSFile(file_path, 'wb') as f:
           pa.ipc.write_stream(table, f)
   

3. 读取时恢复结构

   def load_small_df(file_path):
       # 加载元数据
       with open('df_metadata.json', 'r') as f:
           metadata = json.load(f)
       # 读取数据并恢复原始结构
       with pa.OSFile(file_path, 'rb') as f:
           table = pa.ipc.read_stream(f).to_pandas()
       table = table.astype(metadata['dtypes'])
       if metadata['index_names']:
           table = table.set_index(metadata['index_names'])
       return table
   

方案3：Parquet批次存储（折中） #

将多个小DataPack打包到单个Parquet文件中，用row group区分每个小DataFrame，既减少元数据重复，又保留单文件读取的灵活性：

操作步骤： #

1. 批次写入小DataFrame

   def save_batch_dfs(df_list, batch_file_path):
       # 给每个小DF添加唯一标识，方便后续过滤
       for idx, df in enumerate(df_list):
           df['df_id'] = idx
       combined_df = pd.concat(df_list, ignore_index=False)
       # 每个row group对应一个小DF（300行）
       pq.write_table(pa.Table.from_pandas(combined_df), 
                      batch_file_path, 
                      row_group_size=300,
                      engine='pyarrow')
   

2. 读取单个小DataFrame

   def load_single_df(batch_file_path, df_id):
       # 利用Parquet的row group pruning特性，仅加载目标数据
       table = pq.read_table(batch_file_path, filters=[('df_id', '=', df_id)])
       return table.to_pandas().drop('df_id', axis=1)
   

关键注意事项 #

• 所有小DataFrame必须保证结构完全一致（列名、数据类型、索引），否则元数据复用会失败。
• 若需要后续修改Schema，元数据复用方案会受限，需提前评估需求。

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