使用Polars读取S3多Schema分区Parquet文件报错，求mergeSchema替代方案

Polars目前没有像Spark那样直接的mergeSchema参数，遇到Schema不一致的Parquet文件时，需要手动处理合并逻辑，以下是可行的解决方案：

步骤1：获取所有S3上的Parquet文件路径 #

首先遍历S3路径，拿到所有目标Parquet文件的完整路径，这里用s3fs实现（也可使用boto3）：

import s3fs
import polars as pl

# 初始化S3文件系统
fs = s3fs.S3FileSystem(
    key=access_key,
    secret=secret_key,
    token=token
)

# 匹配所有目标Parquet文件
file_paths = [f"s3://{path}" for path in fs.glob("bucket_name/rs_tables/*/*/*/*.parquet")]

步骤2：合并所有文件的Schema #

遍历所有文件，收集各自的Schema，合并成一个包含所有字段的统一Schema，同时处理类型兼容问题：

# 收集所有文件的Schema
all_schemas = []
for path in file_paths:
    schema = pl.scan_parquet(path, storage_options=storage_options).schema
    all_schemas.append(schema)

# 合并Schema：保留所有字段，自动兼容类型（比如Int32和Int64合并为Int64）
merged_schema = {}
for schema in all_schemas:
    for col_name, dtype in schema.items():
        if col_name not in merged_schema:
            merged_schema[col_name] = dtype
        else:
            # 自动提升兼容类型
            merged_schema[col_name] = pl.datatypes.promote_dtypes([merged_schema[col_name], dtype])

步骤3：统一每个文件的Schema并合并 #

定义函数将单个文件的LazyFrame调整为统一Schema，再合并所有LazyFrame：

def align_schema(lazy_frame):
    current_cols = lazy_frame.columns
    # 添加缺失字段，用Null填充并匹配类型
    for col_name, dtype in merged_schema.items():
        if col_name not in current_cols:
            lazy_frame = lazy_frame.with_columns(pl.lit(None).cast(dtype).alias(col_name))
    # 按统一Schema的顺序和类型输出
    return lazy_frame.select([pl.col(col).cast(dtype) for col, dtype in merged_schema.items()])

# 处理所有文件并合并
processed_lfs = [align_schema(pl.scan_parquet(path, storage_options=storage_options)) for path in file_paths]
final_lazyframe = pl.concat(processed_lfs, how="vertical")

# 执行计算，得到最终DataFrame
final_df = final_lazyframe.collect()

优化建议 #

• 若文件数量极大，可分批次处理后再合并，避免内存过载
• 若字段差异不大，可先抽样部分文件生成合并Schema，减少遍历所有文件的耗时
• 类型合并逻辑可根据业务需求调整，比如某些字段需强制指定类型，可手动修改merged_schema

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