Pandas：合并列类型不一致的Parquet文件，如何用预定义Schema写入？

你的问题核心在于并行导出时，单批次数据的列类型推断不一致（比如某批次全为空值导致列被推断为null类型），进而引发合并读取时的Schema冲突。解决这个问题的关键是提前定义统一的Arrow Schema，强制所有导出文件使用相同的列类型，不管单批次数据是否包含空值。

具体实现步骤 #

1. 定义统一的Arrow Schema #

首先根据你的数据库表结构，用pyarrow.schema定义所有列的类型，确保可能为空的列也明确指定基础类型（比如整数用pa.int64()，字符串用pa.string()等），Arrow类型原生支持空值，无需额外标记。

import pyarrow as pa

# 替换成你的实际列名和类型
unified_schema = pa.schema([
    pa.field("id", pa.int64()),
    pa.field("Deleted?", pa.int64()),  # 明确指定为int64，即使全为null也保留类型
    pa.field("other_column", pa.string())
])

2. 在每个Worker中强制使用该Schema写入Parquet #

在并行导出的Worker函数中，读取数据后将DataFrame转换为Arrow Table，并应用预定义的Schema，再写入Parquet文件。这样不管单批次数据是否有空值，都会按照统一Schema存储。

结合你的复现代码修改示例：

import pandas as pd
import pyarrow.parquet as pq

# 定义统一Schema
schema = pa.schema([
    pa.field("0", pa.int64()),
    pa.field("1", pa.int64())
])

# 写入第一个文件（包含空值）
data = pd.DataFrame([[1, None], [1, None]])
data.columns = data.columns.astype(str)
table = pa.Table.from_pandas(data, schema=schema)
pq.write_table(table, './outputs/1.pq')

# 写入第二个文件（无空值）
data2 = pd.DataFrame([[1, 1], [1, 1]])
data2.columns = data2.columns.astype(str)
table2 = pa.Table.from_pandas(data2, schema=schema)
pq.write_table(table2, './outputs/2.pq')

# 现在读取不会有Schema冲突
dataset = pq.ParquetDataset('./outputs')
result_df = dataset.read_pandas().to_pandas()
print(result_df)

3. 并行场景下的Schema传递 #

在ProcessPool的并行任务中，预定义的Schema可以直接传递给每个Worker（Arrow Schema是可序列化的）：

from multiprocessing import Pool

def export_data_batch(id_range, unified_schema, output_dir):
    # 替换为你的read_sql逻辑：根据id_range读取数据库数据
    query = f"SELECT * FROM your_table WHERE id BETWEEN {id_range[0]} AND {id_range[1]}"
    df = pd.read_sql(query, your_db_connection)
    
    # 确保列名是字符串（Parquet要求）
    df.columns = df.columns.astype(str)
    
    # 转换为Arrow Table并应用统一Schema
    table = pa.Table.from_pandas(df, schema=unified_schema)
    
    # 写入Parquet
    pq.write_table(table, f"{output_dir}/{id_range[0]}.pq")

if __name__ == "__main__":
    # 初始化统一Schema
    unified_schema = pa.schema([
        pa.field("id", pa.int64()),
        pa.field("Deleted?", pa.int64()),
        # 其他列...
    ])
    
    output_dir = "./s3_export_outputs"
    id_ranges = [(1,9), (10,19), (20,29)]
    
    # 启动进程池并行导出
    with Pool(processes=3) as pool:
        pool.starmap(export_data_batch, [(r, unified_schema, output_dir) for r in id_ranges])

备选方案：读取SQL时强制列类型 #

你也可以在pd.read_sql阶段就指定列的dtype，确保每个批次的DataFrame列类型一致，这样写入Parquet时也能避免类型推断不一致：

# 读取SQL时指定nullable整数类型
dtype_mapping = {
    "Deleted?": "Int64"  # pandas的Nullable Int64类型，支持空值
}

df = pd.read_sql(query, your_db_connection, dtype=dtype_mapping)

这种方法适合不想引入Arrow的场景，但用Arrow Schema的方式更贴合Parquet的底层实现，兼容性更好。

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