Build an Automated Reporting System with DuckDB + Airflow

# generate_sales_data.py
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import random

np.random.seed(42)
n_rows = 100000

dates = [datetime(2025, 1, 1) + timedelta(days=random.randint(0, 365)) 
         for _ in range(n_rows)]

categories = ['Electronics', 'Clothing', 'Food', 'Home', 'Sports']
products = {
    'Electronics': ['Phone', 'Laptop', 'Tablet', 'Headphones', 'Watch'],
    'Clothing': ['T-Shirt', 'Jeans', 'Jacket', 'Sneakers', 'Hat'],
    'Food': ['Snacks', 'Beverage', 'Coffee', 'Chocolate', 'Cookies'],
    'Home': ['Lamp', 'Storage Box', 'Carpet', 'Curtains', 'Wall Art'],
    'Sports': ['Yoga Mat', 'Dumbbells', 'Running Shoes', 'Sportswear', 'Water Bottle']
}

data = []
for i in range(n_rows):
    cat = random.choice(categories)
    product = random.choice(products[cat])
    price = round(np.random.exponential(500) + 50, 2)
    quantity = random.randint(1, 10)
    
    data.append({
        'date': dates[i],
        'category': cat,
        'product': product,
        'price': price,
        'quantity': quantity,
        'region': random.choice(['East', 'South', 'North', 'Southwest', 'Northeast']),
        'channel': random.choice(['Online', 'Offline', 'Live'])
    })

df = pd.DataFrame(data)
df.to_parquet('sales_data.parquet', index=False)
print(f'Generated {len(df)} sales records')

pip install pandas numpy pyarrow duckdb
python generate_sales_data.py

# daily_report.py
import duckdb
from datetime import datetime, timedelta

def generate_daily_report():
    con = duckdb.connect()
    con.execute("CREATE VIEW sales AS SELECT * FROM 'sales_data.parquet'")
    
    today = datetime.now().strftime('%Y-%m-%d')
    yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
    
    report = f'=== Daily Report: {today} ===\n\n'
    
    # Overall metrics
    summary = con.execute("""
        SELECT 
            COUNT(*) as total_orders,
            SUM(quantity) as total_units,
            SUM(price * quantity) as total_revenue,
            AVG(price * quantity) as avg_order_value
        FROM sales
        WHERE date = date '{yesterday}'
    """).fetchone()
    
    report += f"""【Overall Metrics】
Total Orders: {summary[0]:,}
Total Units: {summary[1]:,}
Total Revenue: ${summary[2]:,.2f}
Average Order Value: ${summary[3]:,.2f}

【Top 5 Categories】
"""
    
    # Category ranking
    top_categories = con.execute("""
        SELECT 
            category,
            SUM(quantity) as units,
            SUM(price * quantity) as revenue
        FROM sales
        WHERE date = date '{yesterday}'
        GROUP BY category
        ORDER BY revenue DESC
        LIMIT 5
    """).fetchall()
    
    for i, (cat, units, revenue) in enumerate(top_categories, 1):
        report += f"{i}. {cat}: {units:,} units / ${revenue:,.2f}\n"
    
    # Channel distribution
    report += "\n【Channel Distribution】\n"
    channels = con.execute("""
        SELECT 
            channel,
            COUNT(*) as orders,
            SUM(price * quantity) as revenue,
            ROUND(SUM(price * quantity) * 100.0 / SUM(SUM(price * quantity)) OVER(), 2) as pct
        FROM sales
        WHERE date = date '{yesterday}'
        GROUP BY channel
        ORDER BY revenue DESC
    """).fetchall()
    
    for ch, orders, revenue, pct in channels:
        report += f"{ch}: {orders:,} orders / ${revenue:,.2f} ({pct}%)\n"
    
    con.close()
    return report

if __name__ == '__main__':
    report = generate_daily_report()
    print(report)
    
    with open(f'daily_report_{datetime.now().strftime("%Y%m%d")}.txt', 'w') as f:
        f.write(report)
    print('\nReport saved')

# dags/daily_sales_report.py
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator
import sys
sys.path.insert(0, '/path/to/scripts')
from daily_report import generate_daily_report

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'email_on_failure': True,
    'start_date': datetime(2025, 1, 1),
}

with DAG(
    'daily_sales_report',
    default_args=default_args,
    description='Daily Sales Report Auto-Generation',
    schedule_interval='0 8 * * *',
    catchup=False,
    tags=['report', 'sales']
) as dag:
    
    check_data = BashOperator(
        task_id='check_data',
        bash_command='ls -la /data/sales/*.parquet && echo Data ready'
    )
    
    generate_report = PythonOperator(
        task_id='generate_report',
        python_callable=generate_daily_report,
    )
    
    sync_to_cloud = BashOperator(
        task_id='sync_to_cloud',
        bash_command='aws s3 cp daily_report_*.txt s3://company-reports/daily/'
    )
    
    check_data >> generate_report >> sync_to_cloud

con.execute("""
    CREATE OR REPLACE VIEW daily_summary_mv AS
    SELECT 
        date,
        category,
        SUM(price * quantity) as revenue,
        COUNT(*) as orders
    FROM sales
    GROUP BY date, category
""")

con.execute("""
    SELECT * FROM sales 
    WHERE date = ? AND category = ?
""", ['2025-06-15', 'Electronics'])

con.execute("SET threads TO 4")

con.execute("EXPLAIN SELECT * FROM sales WHERE date > '2025-01-01'")