DuckDB 多文件读取与数据质量分析：从 50 个混乱 CSV 到一键洞察

日期,商品名称,类别,数量,单价,销售额,门店
2026-05-01,键盘,外设,10,299,2990,北京店
2026-05-01,鼠标,外设,20,99,1980,北京店
2026-05-02,显示器,硬件,5,2599,12995,北京店

import duckdb

con = duckdb.connect(":memory:")

# 一行代码：读取目录下所有 CSV 并自动合并
result = con.execute("""
    SELECT * FROM read_csv_auto('sales_data/store_*.csv')
""").fetchdf()

print(f"合并了 {len(result)} 条记录")

result = con.execute("""
    SELECT 
        COALESCE("日期", date) AS sale_date,
        COALESCE("商品名称", product) AS product_name,
        COALESCE("类别", category) AS category,
        COALESCE("数量", qty) AS quantity,
        COALESCE("单价", unit_price) AS unit_price,
        COALESCE("销售额", amount) AS amount,
        COALESCE("门店", store) AS store
    FROM read_csv_auto('sales_data/store_*.csv')
""").fetchdf()

import duckdb
from pathlib import Path
from typing import Optional

def merge_csv_glob(
    pattern: str,
    output_format: str = "dataframe",
    columns: Optional[dict] = None
):
    """
    合并 glob 模式匹配的所有 CSV 文件
    
    Args:
        pattern: glob 模式，如 'sales_data/*.csv'
        output_format: 'dataframe', 'table', 或 'parquet'
        columns: 列名映射字典，{'旧名': '新名'}
    
    Returns:
        pandas DataFrame 或 None
    """
    con = duckdb.connect(":memory:")
    
    base_query = f"SELECT * FROM read_csv_auto('{pattern}')"
    
    if columns:
        col_mapping = ", ".join(
            f"COALESCE({old!r}, {new!r}) AS {new}" 
            for old, new in columns.items()
        )
        query = f"SELECT {col_mapping} FROM ({base_query})"
    else:
        query = base_query
    
    if output_format == "dataframe":
        return con.execute(query).fetchdf()
    elif output_format == "parquet":
        output_file = pattern.replace('*', 'merged_output')
        con.execute(f"COPY ({query}) TO '{output_file}' (FORMAT PARQUET)")
        return None
    
    con.close()


# 使用示例
df = merge_csv_glob(
    'sales_data/store_*.csv',
    output_format='dataframe',
    columns={
        '日期': 'sale_date',
        '商品名称': 'product_name',
        '类别': 'category',
        '数量': 'quantity',
        '单价': 'unit_price',
        '销售额': 'amount',
        '门店': 'store'
    }
)

print(f"合并完成：{len(df)} 条记录")

ranking_sql = """
SELECT 
    store,
    COUNT(*) AS transaction_count,
    ROUND(SUM(amount), 2) AS total_sales,
    ROUND(AVG(amount), 2) AS avg_order_value,
    ROUND(SUM(amount) * 100.0 / SUM(SUM(amount)) OVER (), 1) AS share_pct
FROM read_csv_auto('sales_data/store_*.csv')
GROUP BY store
ORDER BY total_sales DESC
LIMIT 10;
"""

con.execute(ranking_sql).print()

spike_sql = """
WITH daily_sales AS (
    SELECT 
        COALESCE("日期", date) AS sale_date,
        ROUND(SUM(amount), 2) AS daily_total
    FROM read_csv_auto('sales_data/store_*.csv')
    GROUP BY 1
),
with_stats AS (
    SELECT 
        sale_date,
        daily_total,
        AVG(daily_total) OVER () AS avg_daily,
        STDDEV_SAMP(daily_total) OVER () AS std_daily
    FROM daily_sales
)
SELECT 
    sale_date,
    daily_total,
    ROUND((daily_total - avg_daily) / NULLIF(std_daily, 0), 2) AS z_score,
    CASE 
        WHEN z_score > 2 THEN '异常高峰'
        WHEN z_score < -2 THEN '异常低谷'
        ELSE '正常'
    END AS status
FROM with_stats
WHERE ABS(z_score) > 2
ORDER BY z_score DESC;
"""

con.execute(spike_sql).print()

category_sql = """
SELECT 
    COALESCE("类别", category) AS category,
    COUNT(DISTINCT COALESCE("门店", store)) AS store_count,
    ROUND(SUM(COALESCE("数量", qty) * COALESCE("单价", unit_price)), 2) AS gross_revenue,
    ROUND(AVG(COALESCE("数量", qty) * COALESCE("单价", unit_price)), 2) AS avg_transaction,
    ROUND(100.0 * SUM(COALESCE("数量", qty) * COALESCE("单价", unit_price)) / 
          NULLIF(SUM(COALESCE("数量", qty) * COALESCE("单价", unit_price)) OVER (), 0), 1) AS revenue_share
FROM read_csv_auto('sales_data/store_*.csv')
GROUP BY 1
ORDER BY gross_revenue DESC;
"""

con.execute(category_sql).print()

# 检查缺失值分布
quality_sql = """
SELECT 
    'NULL_COUNT' AS metric,
    SUM(CASE WHEN amount IS NULL THEN 1 ELSE 0 END) AS null_count,
    SUM(CASE WHEN quantity IS NULL THEN 1 ELSE 0 END) AS qty_null,
    SUM(CASE WHEN store IS NULL THEN 1 ELSE 0 END) AS store_null,
    COUNT(*) AS total_rows
FROM read_csv_auto('sales_data/store_*.csv')

UNION ALL

SELECT 
    'VALUE_RANGE' AS metric,
    MIN(amount)::VARCHAR,
    MAX(amount)::VARCHAR,
    MIN(quantity)::VARCHAR,
    MAX(quantity)::VARCHAR
FROM read_csv_auto('sales_data/store_*.csv');
"""

con.execute(quality_sql).print()

# 直接写入 Parquet——文件更小、读取更快
con.execute("""
    COPY (
        SELECT 
            COALESCE("日期", date) AS sale_date,
            COALESCE("商品名称", product) AS product_name,
            COALESCE("类别", category) AS category,
            COALESCE("数量", qty) AS quantity,
            COALESCE("单价", unit_price) AS unit_price,
            COALESCE("销售额", amount) AS amount,
            COALESCE("门店", store) AS store
        FROM read_csv_auto('sales_data/store_*.csv')
    ) TO 'merged_all_stores.parquet' (FORMAT PARQUET);
""")

import os
print(f"Parquet 文件大小: {os.path.getsize('merged_all_stores.parquet') / 1024:.1f} KB")