DuckDB 联邦查询实战：不用搬数据，一行 SQL 查遍所有数据库

import duckdb

conn = duckdb.connect()

# 挂载 PostgreSQL 数据库
conn.execute("ATTACH 'dbname=production host=localhost user=postgres' AS pg (READ_ONLY)")

# 挂载 MySQL 数据库
conn.execute("ATTACH 'host=localhost port=3306 user=root password=secret dbname=logs' AS mysql (READ_ONLY, TYPE mysql)")

# 挂载 SQLite 数据库
conn.execute("ATTACH 'users.db' AS sqlite (READ_ONLY)")

# 直接查询远程 PostgreSQL（不需要 ATTACH）
result = conn.execute("""
    SELECT * FROM read_parquet('s3://bucket/data/*.parquet')
""").fetchdf()

# 查询 HTTP API 返回的 JSON
result = conn.execute("""
    SELECT * FROM read_json_auto('https://api.example.com/data.json')
""").fetchdf()

import duckdb

conn = duckdb.connect()

# 挂载 PostgreSQL（只读模式，确保不会意外修改生产库）
conn.execute("ATTACH 'dbname=production host=localhost user=postgres' AS pg (READ_ONLY)")

# 跨库查询：订单 + 用户信息
result = conn.execute("""
    SELECT 
        p.name, 
        SUM(o.amount) as total_sales,
        COUNT(DISTINCT o.id) as order_count
    FROM pg.public.orders o
    JOIN pg.public.customers c ON o.customer_id = c.id
    WHERE o.created_at >= '2026-07-01'
    GROUP BY p.name
    ORDER BY total_sales DESC
    LIMIT 10
""").fetchdf()

print(result)

# 挂载 MySQL
conn.execute("""
    ATTACH 'host=localhost port=3306 user=root password=secret dbname=logs' 
    AS mysql (READ_ONLY, TYPE mysql)
""")

# 查询日志数据
result = conn.execute("""
    SELECT 
        DATE(created_at) as day, 
        COUNT(*) as events,
        COUNT(DISTINCT user_id) as active_users
    FROM mysql.public.events
    WHERE created_at >= '2026-08-01'
    GROUP BY DATE(created_at)
    ORDER BY day
""").fetchdf()

print(result)

# 挂载 SQLite（无需额外安装驱动）
conn.execute("ATTACH 'users.db' AS sqlite (READ_ONLY)")

# 查询用户数据
result = conn.execute("""
    SELECT * FROM sqlite.users 
    WHERE active = true 
    ORDER BY created_at DESC
    LIMIT 100
""").fetchdf()

# 假设：
# - pg.public.users：PostgreSQL 中的用户表
# - sqlite.sessions：SQLite 中的会话记录
# - sales/*.parquet：Parquet 格式的销售数据

result = conn.execute("""
    SELECT 
        p.name,
        s.total_sales,
        COUNT(sl.id) as login_count,
        AVG(sl.duration) as avg_session_duration
    FROM pg.public.users p
    JOIN sqlite.sessions sl ON p.id = sl.user_id
    JOIN read_parquet('sales/*.parquet') s ON p.id = s.user_id
    GROUP BY p.name, s.total_sales
    ORDER BY s.total_sales DESC
""").fetchdf()

print(result)

# ✅ 正确：始终使用 READ_ONLY
conn.execute("ATTACH 'dbname=production host=localhost' AS pg (READ_ONLY)")

# ❌ 危险：忘记加 READ_ONLY 可能误改生产数据
conn.execute("ATTACH 'dbname=production host=localhost' AS pg")

# ✅ 正确：先过滤再 JOIN，减少数据传输
result = conn.execute("""
    SELECT * FROM pg.public.orders 
    WHERE created_at >= '2026-08-01'
""").fetchdf()

result2 = conn.execute("""
    SELECT * FROM sqlite.users 
    WHERE active = true
""").fetchdf()

# 在 DuckDB 内存中 JOIN
final = conn.execute("""
    SELECT * FROM result2 
    JOIN result ON result2.id = result.user_id
""").fetchdf()

# ❌ 危险：直接跨库 JOIN 大表，网络传输量大
result = conn.execute("""
    SELECT * FROM pg.public.orders o
    JOIN sqlite.users u ON o.user_id = u.id
""").fetchdf()

# ✅ 正确：复用 Connection 对象
conn = duckdb.connect()
conn.execute("ATTACH '...' AS pg (READ_ONLY)")
conn.execute("ATTACH '...' AS mysql (READ_ONLY)")

# 多次查询复用同一连接
for date_range in date_ranges:
    result = conn.execute(f"""
        SELECT * FROM pg.public.orders 
        WHERE created_at BETWEEN '{date_range[0]}' AND '{date_range[1]}'
    """).fetchdf()

# ❌ 错误：每次创建新连接
for date_range in date_ranges:
    conn = duckdb.connect()  # 重新连接，性能差
    conn.execute("ATTACH '...' AS pg (READ_ONLY)")
    ...

import duckdb
import schedule
import time

def daily_price_analysis():
    conn = duckdb.connect()
    
    # 挂载各数据源
    conn.execute("ATTACH 'production.db' AS pg (READ_ONLY)")
    conn.execute("ATTACH 'competitor_prices.csv' AS comp (TYPE CSV)")
    
    # 分析竞品价格变动
    result = conn.execute("""
        SELECT 
            c.product_name,
            c.current_price,
            p.avg_price as our_price,
            ROUND((c.current_price - p.avg_price) / p.avg_price * 100, 2) as price_gap_pct
        FROM comp.competitor_prices c
        LEFT JOIN pg.public.our_prices p ON c.product_id = p.product_id
        WHERE c.updated_at >= CURRENT_DATE - INTERVAL '7' DAY
        ORDER BY price_gap_pct ASC
    """).fetchdf()
    
    # 输出报告
    print("📊 竞品价格分析报告")
    print(result.to_string(index=False))
    
    # 导出 JSON 供下游使用
    result.to_json("price_report.json", orient='records', indent=2)

# 每天早上 8 点执行
schedule.every().day.at("08:00").do(daily_price_analysis)

while True:
    schedule.run_pending()
    time.sleep(60)