背景：周报这件事，每天都在消耗数据人的时间
每个数据团队每周都要写周报。你手动导出数据、写图表、整理结论，一周花 2-3 小时。如果给多个业务线写，这个时间还会翻倍。
我们用 DuckDB + Python 搭一个全自动周报系统：每天凌晨自动跑，产出报告，发到 Telegram。你自己省时间，还能把这个能力卖给其他团队。
第一步：准备测试数据
假设你有一份销售数据（CSV 格式）：
date,product,region,revenue
2024-01-01,iPhone,华东,120000
2024-01-01,iPad,华北,85000
创建一个虚拟 CSV 用于测试：
import os
os.makedirs('/tmp/duckdb_weekly', exist_ok=True)
with open('/tmp/duckdb_weekly/sales.csv', 'w') as f:
import random
products = ['iPhone', 'iPad', 'MacBook', 'AirPods']
regions = ['华东', '华北', '华南', '西南']
f.write('date,product,region,revenue\n')
for day in range(1, 32):
for _ in range(50):
f.write(f'2024-01-{day:02d},{random.choice(products)},{random.choice(regions)},{random.randint(5000, 50000)}\n')
print("数据生成完毕")
第二步：核心查询——生成周报
DuckDB 的 SQL 语法直接解决 80% 的问题：
import duckdb
from datetime import datetime, timedelta
conn = duckdb.connect()
conn.execute("CREATE TABLE sales AS SELECT * FROM read_csv('/tmp/duckdb_weekly/sales.csv')")
end_date = datetime(2024, 1, 31)
start_date = end_date - timedelta(days=6)
weekly_report = conn.execute("""
SELECT
DATE(date) AS report_date,
product,
region,
SUM(revenue) AS total_revenue,
COUNT(*) AS order_count,
AVG(revenue) AS avg_order_value
FROM sales
WHERE date BETWEEN ? AND ?
GROUP BY 1, 2, 3
ORDER BY total_revenue DESC
""", str(start_date.date()), str(end_date.date())).fetchdf()
print(weekly_report.head(10))
这个查询的输出就是周报的核心数据：按产品、地区分组的销售额、订单数、客单价。
第三步：生成可视化
import pandas as pd
import matplotlib.pyplot as plt
product_summary = weekly_report.groupby('product')['total_revenue'].sum().sort_values(ascending=False)
plt.figure(figsize=(10, 6))
product_summary.plot(kind='bar', color='steelblue')
plt.title('Weekly Revenue by Product')
plt.ylabel('Revenue (CNY)')
plt.xlabel('Product')
plt.tight_layout()
plt.savefig('/tmp/duckdb_weekly/chart.png', dpi=150)
print("图表已生成")
第四步：打包成 Telegram 消息
import telebot
BOT_TOKEN='***'
CHAT_ID = 'YOUR_CHAT_ID'
bot = telebot.TeleBot(BOT_TOKEN)
message = f"""
📊 周报已生成 (2024-01-25 至 2024-01-31)
🏆 Top 产品：
{weekly_report.head(3).to_string(index=False)}
📈 本周总营收：{weekly_report['total_revenue'].sum():,.0f} 元
📦 总订单数：{weekly_report['order_count'].sum():,}
💰 平均客单价：{weekly_report['avg_order_value'].mean():,.0f} 元
[查看完整报告附件]
"""
bot.send_photo(CHAT_ID, open('/tmp/duckdb_weekly/chart.png', 'rb'), caption=message)
print("周报已发送到 Telegram")
第五步：自动化调度
用 cron 定时任务，每天早上 8 点自动执行：
crontab -e
# Add this line (runs daily at 8:00 AM)
0 8 * * * cd /root && python3 weekly_report.py >> /tmp/duckdb_weekly/cron.log 2>&1
或者用 Python 的 APScheduler 库，更灵活：
from apscheduler.schedulers.background import BackgroundScheduler
import time
def run_weekly():
print(f"[{datetime.now()}] 开始生成周报...")
# 调用上面的生成逻辑
print("周报生成完成")
scheduler = BackgroundScheduler()
scheduler.add_job(run_weekly, 'cron', day_of_week='mon-fri', hour=8, minute=0)
scheduler.start()
while True:
time.sleep(60)
进阶：多数据源 + Parquet 缓存
生产环境的数据通常不止一个 CSV。DuckDB 支持直接读 Parquet、SQLite、甚至远程 HTTP 接口：
# 读取 Parquet（预处理过的数据，查询秒级响应）
conn.execute("CREATE TABLE sales_pq AS SELECT * FROM read_parquet('/data/sales/*.parquet')")
# 读取远程 CSV（通过 httpfs 扩展）
conn.execute("INSTALL httpfs; LOAD httpfs;")
conn.execute("CREATE TABLE remote_sales AS SELECT * FROM read_csv_auto('https://api.example.com/sales.csv')")
# 跨表 JOIN
conn.execute("""
SELECT s.product, s.region, SUM(s.revenue) as revenue
FROM sales_pq s
JOIN remote_sales r ON s.order_id = r.id
GROUP BY 1, 2
""")
进阶：Web 界面——Streamlit 实时看板
import streamlit as st
import duckdb
st.title('实时周报看板')
df = duckdb.query("SELECT * FROM read_csv('sales.csv')").df()
st.dataframe(df)
st.line_chart(df.set_index('date')['revenue'])
变现路径：从工具到产品
- 内部工具：自己团队用，省时间 → 每小时节省 2 元 × 52 周 = 年省 10 万+
- SaaS 产品：做成多租户服务，99 元/月/企业 → 100 家 = 年入 12 万
- 外包服务：帮中小企业搭建，单项目收费 2000-5000 元
- 开源项目引流：把代码开源在 GitHub，靠咨询和培训变现
传统方案 vs DuckDB 方案对比
传统方案需要 MySQL + Airflow + Jupyter + 邮件服务，开发周期长、维护成本高。DuckDB 方案一个脚本搞定，零运维。列式存储让聚合计算比 pandas 快 10 倍以上，且无需安装任何数据库服务。
本文的完整可运行代码和项目模板已发布在 duckdblab.org，包含更详细的步骤和更多案例。想系统学习 DuckDB 在商业场景的应用，duckdblab.org 上有完整的教程系列，从入门到实战一网打尽。
学习更多 DuckDB 实战技巧 → duckdblab.org