用 DuckDB MERGE INTO 干掉 ETL 里的 Upsert 噩梦：一行 SQL 替代 20 行 Python 代码

# 传统做法：检查是否存在，然后决定 INSERT 还是 UPDATE
existing = con.execute("SELECT id FROM daily_sales WHERE date = '2026-06-30'").fetchall()
if existing:
    con.execute("UPDATE daily_sales SET ... WHERE date = '2026-06-30'")
else:
    con.execute("INSERT INTO daily_sales VALUES (...)")

MERGE INTO target_table
USING source_data
ON target_table.key = source_data.key
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT VALUES ...

import duckdb

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

# 主表：按订单号去重
con.execute("""
    CREATE TABLE daily_sales (
        order_id VARCHAR PRIMARY KEY,
        product_id VARCHAR,
        quantity INTEGER,
        amount DOUBLE,
        sale_date DATE,
        region VARCHAR,
        updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

# 临时表：存放当天消费的全部原始数据（可能含重复）
con.execute("""
    CREATE TEMP TABLE raw_ingest AS
    SELECT * FROM read_csv_auto('/data/kafka/sales_20260630.csv')
""")

# 核心：一行 SQL 搞定幂等写入
con.execute("""
    MERGE INTO daily_sales ds
    USING raw_ingest ri
    ON ds.order_id = ri.order_id
    WHEN MATCHED THEN
        UPDATE SET
            quantity = ds.quantity + ri.quantity,
            amount = ds.amount + ri.amount,
            updated_at = CURRENT_TIMESTAMP
    WHEN NOT MATCHED THEN
        INSERT (order_id, product_id, quantity, amount, sale_date, region)
        VALUES (ri.order_id, ri.product_id, ri.quantity, ri.amount, ri.sale_date, ri.region)
""")

# 验证结果
result = con.execute("""
    SELECT
        COUNT(*) as total_records,
        COUNT(DISTINCT order_id) as unique_orders,
        SUM(quantity) as total_qty,
        ROUND(SUM(amount), 2) as total_revenue
    FROM daily_sales
""").fetchall()

print(f'总记录: {result[0][0]} | 唯一订单: {result[0][1]} | 总收入: {result[0][3]}')

-- 只更新金额变化超过 10% 的记录
MERGE INTO daily_sales ds
USING raw_ingest ri
ON ds.order_id = ri.order_id
WHEN MATCHED AND ABS(ds.amount - ri.amount) / NULLIF(ds.amount, 0) > 0.1 THEN
    UPDATE SET
        amount = ri.amount,
        quantity = ri.quantity,
        updated_at = CURRENT_TIMESTAMP
WHEN NOT MATCHED THEN
    INSERT (order_id, product_id, quantity, amount, sale_date, region)
    VALUES (ri.order_id, ri.product_id, ri.quantity, ri.amount, ri.sale_date, ri.region);

-- 在 MERGE 之前先去重，减少匹配工作量
CREATE TEMP TABLE deduped_raw AS
SELECT 
    order_id,
    product_id,
    SUM(quantity) as quantity,
    SUM(amount) as amount,
    MIN(sale_date) as sale_date,
    FIRST_VALUE(region) OVER (PARTITION BY order_id) as region
FROM raw_ingest
GROUP BY order_id, product_id, sale_date

SET threads = 0;  -- 使用所有 CPU 核心

# MERGE 完成后，导出为 Parquet 供后续分析
con.execute("""
    COPY daily_sales TO '/data/daily/daily_sales_20260630.parquet'
    (FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 100000)
""")

-- DuckDB 不直接返回 MERGE 影响的行数，但可以事后验证
SELECT 
    (SELECT COUNT(*) FROM daily_sales) as current_total,
    (SELECT COUNT(*) FROM raw_ingest) as source_total,
    (SELECT COUNT(DISTINCT order_id) FROM raw_ingest) as source_unique;