用 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 = ['电子', '服装', '食品', '家居', '运动']
products = {
    '电子': ['手机', '笔记本', '平板', '耳机', '手表'],
    '服装': ['T恤', '牛仔裤', '外套', '运动鞋', '帽子'],
    '食品': ['零食', '饮料', '咖啡', '巧克力', '饼干'],
    '家居': ['台灯', '收纳盒', '地毯', '窗帘', '装饰画'],
    '运动': ['瑜伽垫', '哑铃', '跑步鞋', '运动服', '水壶']
}

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(['华东', '华南', '华北', '西南', '东北']),
        'channel': random.choice(['线上', '线下', '直播'])
    })

df = pd.DataFrame(data)
df.to_parquet('sales_data.parquet', index=False)
print(f'已生成 {len(df)} 条销售数据')

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'=== 日报：{today} ===\n\n'
    
    # 整体指标
    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"""【整体指标】
总订单数: {summary[0]:,}
总销量: {summary[1]:,} 件
总营收: ¥{summary[2]:,.2f}
平均客单价: ¥{summary[3]:,.2f}

【品类 TOP5】
"""
    
    # 品类排名
    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:,}件 / ¥{revenue:,.2f}\n"
    
    # 渠道分布
    report += "\n【渠道分布】\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:,}单 / ¥{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('\n报告已保存')

# 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='每日销售报表自动生成',
    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', '电子'])

con.execute("SET threads TO 4")

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