DuckDB Federated Query in Practice: MySQL + CSV One-Stop Aggregation

pip install duckdb pandas python-dotenv sqlalchemy pymysql

# config.py
import os
from dotenv import load_dotenv

load_dotenv()

MYSQL_DSN = os.getenv("MYSQL_DSN")        # mysql+pymysql://user:pass@host:3306/dbname
EXPORT_DIR = os.getenv("EXPORT_DIR", "./exports")
REPORT_DB = os.getenv("REPORT_DB", "report.duckdb")

import duckdb

conn = duckdb.connect("report.duckdb")

# Load the MySQL extension
conn.execute("INSTALL mysql; LOAD mysql;")

# ATTACH the remote MySQL database
mysql_dsn = "mysql+pymysql://user:password@localhost:3306/ecommerce"
conn.execute(f"ATTACH '{mysql_dsn}' AS mysql_db (TYPE mysql);")

# Now you can query MySQL tables directly
result = conn.execute("""
    SELECT order_id, user_id, amount, channel, created_at
    FROM mysql_db.orders
    WHERE DATE(created_at) = '2026-09-18'
""").fetchdf()

print(f"Yesterday's orders: {len(result)}")

# List all attached databases
conn.execute("SHOW DATABASES;").fetchdf()

# List tables in MySQL
conn.execute("SHOW TABLES FROM mysql_db;").fetchdf()

# Read yesterday's ad spend CSV
ad_df = conn.read_csv_auto("./exports/ad_spend_2026-09-18.csv")

# Or specify column types explicitly to avoid misinference
ad_df = conn.read_csv_auto(
    "./exports/ad_spend_2026-09-18.csv",
    columns={
        "platform": "VARCHAR",
        "spend": "DOUBLE",
        "impressions": "BIGINT",
        "clicks": "INTEGER",
        "date": "DATE"
    }
)

def build_daily_report(conn, date_str):
    """Cross-source daily report: MySQL orders + CSV ad spend"""

    # Register CSV as a temporary view
    ad_df = conn.read_csv_auto(f"./exports/ad_{date_str}.csv")
    conn.register("ad_spend", ad_df)

    # Core query: JOIN MySQL orders + CSV ad spend
    report_sql = f"""
    WITH order_stats AS (
        SELECT
            channel,
            COUNT(*) AS total_orders,
            SUM(amount) AS total_gmv,
            ROUND(SUM(amount) / COUNT(*), 2) AS avg_order_value
        FROM mysql_db.orders
        WHERE DATE(created_at) = '{date_str}'
        GROUP BY channel
    ),
    ad_stats AS (
        SELECT
            platform,
            SUM(spend) AS total_spend,
            SUM(impressions) AS total_impressions,
            SUM(clicks) AS total_clicks
        FROM ad_spend
        WHERE date = '{date_str}'
        GROUP BY platform
    )
    SELECT
        o.channel AS metric_channel,
        o.total_orders,
        o.total_gmv,
        o.avg_order_value,
        a.total_spend,
        a.total_impressions,
        a.total_clicks,
        ROUND(a.total_spend * 1.0 / NULLIF(o.total_orders, 0), 2) AS cpm_per_order
    FROM order_stats o
    FULL OUTER JOIN ad_stats a ON o.channel = a.platform
    ORDER BY o.total_gmv DESC
    """

    report = conn.execute(report_sql).fetchdf()
    return report

# The orders table has a json_data field storing cart items
json_query = """
    SELECT
        order_id,
        amount,
        channel,
        json_array_length(json_data::JSON->'items') AS item_count,
        json_extract_string(json_data::JSON->'items', '$[0].name') AS first_item
    FROM mysql_db.orders
    WHERE DATE(created_at) = '2026-09-18'
    LIMIT 100
"""
sample = conn.execute(json_query).fetchdf()

import schedule
import time
from datetime import datetime, timedelta

class DailyReportEngine:
    def __init__(self):
        self.conn = duckdb.connect("report.duckdb")
        self.conn.execute("INSTALL mysql; LOAD mysql;")
        self.conn.execute(
            "ATTACH 'mysql+pymysql://user:pass@host:3306/ecommerce' AS mysql_db (TYPE mysql);"
        )

    def run(self):
        yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
        report = self.build_daily_report(yesterday)

        # Save as CSV
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        report.to_csv(f"./exports/report_{ts}.csv", index=False)

        # Send notification (DingTalk/Feishu/Webhook)
        self.notify(report)
        print(f"✅ Report generated: {ts}")

    def notify(self, report):
        print(report.to_string())

if __name__ == "__main__":
    engine = DailyReportEngine()
    schedule.every().day.at("08:30").do(engine.run)
    print("⏰ Scheduled task started, runs daily at 08:30")
    while True:
        schedule.run_pending()
        time.sleep(60)

# Edit crontab
crontab -e

# Auto-run daily at 8:30 AM
30 8 * * * cd /home/user/project && /usr/bin/python3 daily_report.py >> /var/log/duckdb_report.log 2>&1