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'")