DuckDB Parquet Schema 一致性完全指南：解决列顺序不一致的读取报错

-- 基础用法：自动按列名对齐
SELECT * 
FROM read_parquet('data/*.parquet', union_by_name=true);

-- 显式指定需要的列（更安全，避免意外引入脏数据）
SELECT date, user_id, amount, category
FROM read_parquet('data/*.parquet', union_by_name=true);

-- 结合 hive_partitioning 处理分区目录
SELECT *
FROM read_parquet('data/year=2026/*.parquet', 
                  hive_partitioning=true,
                  union_by_name=true);

import duckdb

# 读取所有 Parquet 文件，自动对齐列
df = duckdb.sql("""
    SELECT * FROM read_parquet('data/*.parquet', 
                                union_by_name=true)
""").df()

# 只选择需要的列
df = duckdb.sql("""
    SELECT date, user_id, amount, category
    FROM read_parquet('data/*.parquet', union_by_name=true)
""").df()

# 结合分区路径
df = duckdb.sql("""
    SELECT * FROM read_parquet('data/year=2026/*.parquet', 
                                hive_partitioning=true,
                                union_by_name=true)
""").df()

-- 明确指定列，DuckDB 自动补齐缺失列
SELECT 
    date, 
    user_id, 
    amount, 
    category
FROM read_parquet('data/*.parquet', union_by_name=true)
WHERE date >= DATE '2026-01-01'
  AND amount > 0;

import pyarrow as pa
import pyarrow.parquet as pq
from datetime import date

# 定义统一 schema
schema = pa.schema([
    ('date', pa.date32()),
    ('user_id', pa.int64()),
    ('amount', pa.float64()),
    ('category', pa.string())
])

# 写入时强制使用同一 schema
table = pa.table({
    'date': [pa.scalar(date(2026, 8, 22), type=pa.date32())],
    'user_id': [12345],
    'amount': [99.9],
    'category': ['electronics']
}, schema=schema)

pq.write_table(table, 'output.parquet')

# 验证：用 DuckDB 读取无需任何特殊参数
import duckdb
df = duckdb.sql("SELECT * FROM read_parquet('output.parquet')").df()
print(df)

import pyarrow as pa
import pyarrow.parquet as pq
import glob
from pathlib import Path

# 定义目标 schema
TARGET_SCHEMA = pa.schema([
    ('date', pa.date32()),
    ('user_id', pa.int64()),
    ('amount', pa.float64()),
    ('category', pa.string())
])

def normalize_parquet_schema(input_dir, output_dir):
    """批量统一 Parquet 文件 schema"""
    Path(output_dir).mkdir(exist_ok=True)
    
    for filepath in glob.glob(f'{input_dir}/*.parquet'):
        # 读取文件
        table = pq.read_table(filepath)
        
        # 对齐 schema（缺失列补 NULL，多余列丢弃）
        normalized = table.cast(TARGET_SCHEMA, safe=False)
        
        # 写入新文件
        out_path = Path(output_dir) / Path(filepath).name
        pq.write_table(normalized, out_path)
        print(f'Normalized: {filepath} -> {out_path}')

normalize_parquet_schema('raw_data/', 'normalized_data/')

-- 查看单个文件的 schema
DESCRIBE SELECT * FROM read_parquet('data/file.parquet');

-- 批量查看所有文件的 schema 结构
SELECT 
    file_name, 
    column_name, 
    column_type,
    ordinal_position
FROM parquet_schema('data/*.parquet')
ORDER BY file_name, ordinal_position;

-- 找出 schema 不一致的文件
WITH schemas AS (
    SELECT 
        file_name,
        list(zip(column_name, column_type)) as col_types
    FROM parquet_schema('data/*.parquet')
    GROUP BY file_name
)
SELECT 
    a.file_name as file_a,
    b.file_name as file_b,
    a.col_types as schema_a,
    b.col_types as schema_b
FROM schemas a
JOIN schemas b ON a.file_name < b.file_name
WHERE a.col_types != b.col_types;

import duckdb
import fastparquet
import pandas as pd

# 读取 fastparquet 写入的文件，自动提升类型兼容
df = duckdb.read_parquet('fastparquet_output.parquet', promote_types=True)

# 或者在 DuckDB 中直接处理
result = duckdb.sql("""
    SELECT * 
    FROM read_parquet('fastparquet_output.parquet', 
                      promote_types=true)
""").df()

-- DuckDB 中用 promote_types 自动处理类型提升
SELECT * 
FROM read_parquet('data/*.parquet', promote_types=true);

import duckdb
from datetime import datetime

def read_parquet_robust(pattern, required_columns, start_date=None):
    """
    构建鲁棒的 Parquet 读取函数
    
    Args:
        pattern: Parquet 文件路径模式，如 'data/*.parquet'
        required_columns: 必需的列名列表
        start_date: 可选，只读取此日期之后的数据
    
    Returns:
        DuckDB DataFrame
    """
    query = f"""
        SELECT {', '.join(required_columns)}
        FROM read_parquet('{pattern}', 
                          union_by_name=true,
                          hive_partitioning=true)
    """
    
    if start_date:
        query += f" WHERE date >= DATE '{start_date}'"
    
    return duckdb.sql(query)

# 使用示例
df = read_parquet_robust(
    pattern='s3://my-bucket/parquet/2026/*.parquet',
    required_columns=['date', 'user_id', 'amount', 'category'],
    start_date='2026-01-01'
)

# 直接聚合，不加载到内存
result = df.aggregate([
    ('SUM(amount)', 'total_amount'),
    ('COUNT(*)', 'record_count'),
    ('AVG(amount)', 'avg_amount')
])
print(result)