DuckDB 端到端自动化报表流水线：从数据源到邮件推送的全流程实战
工具链搭建 | 适合：每周手动整理报表的数据分析师
痛点场景
你每周一早上要花 3 小时手动整理 Excel，重复、无聊、易出错。如果用 DuckDB 搭一套自动化报表系统，这套流程可以压缩到 5 分钟，甚至全自动推送。
本文带你从零搭建一个完整的 DuckDB + Python 自动化报表流水线，涵盖数据接入、分析查询、格式输出、定时调度和邮件推送五个环节。
系统架构总览
数据源(Excel/CSV/SQL)
↓
DuckDB(统一查询层)
↓
Python 处理(数据清洗/聚合)
↓
输出(Excel/PDF/邮件推送)
核心优势：DuckDB 可以直接读取 Excel 和 CSV 文件，无需中间转换；SQL 表达能力强，复杂聚合一条语句搞定；配合 Python 生态可以串联整个工作流。
第一步：环境准备
pip install duckdb openpyxl pandas
验证安装：
import duckdb
print(duckdb.__version__) # 应该输出版本号
第二步：连接数据源
假设你有一个电商销售数据文件夹 sales_data/，包含多个 CSV 文件，每个文件是一个月的销售记录。
import duckdb
import pandas as pd
from pathlib import Path
# 连接 DuckDB 持久化数据库（数据会保存到磁盘）
con = duckdb.connect('sales_automation.duckdb')
# 批量读取所有 CSV（自动合并，无需逐个读取）
con.execute("""
CREATE TABLE all_sales AS
SELECT * FROM read_csv_auto('sales_data/*.csv', hive_partitioning=true)
""")
# 查看数据概览
result = con.execute("DESCRIBE all_sales").fetchdf()
print(result)
关键技巧：
read_csv_auto会自动推断列类型，不需要手动指定 schemahive_partitioning=true支持按目录结构自动分区，文件名中的日期会被提取为分区列- DuckDB 的列式存储让聚合查询比 Pandas 快 10 倍以上
第三步：核心分析查询
3.1 月度销售总览
monthly_summary = con.execute("""
SELECT
DATE_TRUNC('month', sale_date) AS month,
SUM(revenue) AS total_revenue,
COUNT(DISTINCT customer_id) AS active_customers,
AVG(order_value) AS avg_order_value,
SUM(CASE WHEN region = '华东' THEN revenue ELSE 0 END) AS east_china_revenue
FROM all_sales
GROUP BY DATE_TRUNC('month', sale_date)
ORDER BY month DESC
""").fetchdf()
print(monthly_summary)
3.2 TOP 商品排行（含窗口函数）
top_products = con.execute("""
SELECT
product_name,
category,
SUM(quantity) AS total_qty,
SUM(revenue) AS total_revenue,
RANK() OVER (ORDER BY SUM(revenue) DESC) AS revenue_rank
FROM all_sales
GROUP BY product_name, category
HAVING SUM(revenue) > 1000
ORDER BY total_revenue DESC
LIMIT 20
""").fetchdf()
3.3 客户分层分析
customer_segment = con.execute("""
SELECT
customer_id,
COUNT(*) AS order_count,
SUM(revenue) AS total_spend,
MAX(sale_date) AS last_purchase,
CASE
WHEN SUM(revenue) > 10000 THEN 'VIP'
WHEN SUM(revenue) > 5000 THEN '高价值'
WHEN SUM(revenue) > 1000 THEN '普通'
ELSE '潜力'
END AS segment
FROM all_sales
GROUP BY customer_id
ORDER BY total_spend DESC
""").fetchdf()
第四步：生成格式化报表
4.1 Excel 格式（带样式）
import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment
def generate_excel_report(df_dict, output_path='report.xlsx'):
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
for sheet_name, df in df_dict.items():
df.to_excel(writer, sheet_name=sheet_name, index=False)
# 美化表头
ws = writer.sheets[sheet_name]
header_fill = PatternFill(start_color='4472C4', end_color='4472C4', fill_type='solid')
header_font = Font(bold=True, color='FFFFFF')
for cell in ws[1]:
cell.fill = header_fill
cell.font = header_font
cell.alignment = Alignment(horizontal='center')
# 自动调整列宽
for column in ws.columns:
max_length = 0
column_letter = column[0].column_letter
for cell in column:
try:
if len(str(cell.value)) > max_length:
max_length = len(str(cell.value))
except:
pass
adjusted_width = min(max_length + 2, 50)
ws.column_dimensions[column_letter].width = adjusted_width
print(f'报表已生成：{output_path}')
# 生成多 Sheet 报表
report_data = {
'月度汇总': monthly_summary,
'TOP商品': top_products,
'客户分层': customer_segment
}
generate_excel_report(report_data, 'weekly_report.xlsx')
4.2 纯文本格式（简洁版）
def generate_text_report(df_dict, output_path='report.txt'):
with open(output_path, 'w', encoding='utf-8') as f:
f.write('=' * 60 + '\n')
f.write('周报数据报告\n')
f.write('=' * 60 + '\n\n')
for sheet_name, df in df_dict.items():
f.write(f'\n【{sheet_name}】\n')
f.write(df.to_string(index=False))
f.write('\n' + '-' * 60 + '\n')
print(f'文本报告已生成：{output_path}')
第五步：定时调度（三种方案任选）
方案 A：Python schedule 库（适合本地运行）
import schedule
import time
from datetime import datetime
def run_weekly_report():
print(f'{datetime.now()} - 开始生成周报...')
# 重新加载最新数据
con.execute("DROP TABLE IF EXISTS all_sales")
con.execute("""
CREATE TABLE all_sales AS
SELECT * FROM read_csv_auto('sales_data/*.csv', hive_partitioning=true)
""")
# 执行分析查询
monthly_summary = con.execute("""
SELECT DATE_TRUNC('month', sale_date) AS month,
SUM(revenue) AS total_revenue, COUNT(DISTINCT customer_id) AS active_customers
FROM all_sales GROUP BY DATE_TRUNC('month', sale_date)
ORDER BY month DESC
""").fetchdf()
# 输出报表
report_data = {'月度汇总': monthly_summary}
generate_excel_report(report_data, f'report_{datetime.now().strftime("%Y%m%d")}.xlsx')
print('周报生成完成！')
# 每周一早上 9 点执行
schedule.every().monday.at("09:00").do(run_weekly_report)
while True:
schedule.run_pending()
time.sleep(60)
方案 B：系统 Cron（Linux/Mac）
crontab -e
添加以下行（每周一 9:00 执行）：
0 9 * * 1 cd /path/to/project && python3 weekly_report.py
方案 C：GitHub Actions（云端免费调度）
创建 .github/workflows/report.yml：
name: Weekly Report
on:
schedule:
- cron: '0 9 * * 1' # 周一 9:00 UTC
workflow_dispatch:
jobs:
build-report:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: pip install duckdb openpyxl pandas
- name: Generate report
run: python3 weekly_report.py
- name: Upload artifact
uses: actions/upload-artifact@v3
with:
name: weekly-report
path: report_*.xlsx
第六步：邮件推送（可选）
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
def send_email_report(to_email, report_path, subject='周报数据报告'):
msg = MIMEMultipart()
msg['From'] = '[email protected]'
msg['To'] = to_email
msg['Subject'] = subject
body = '附件为本周数据报告，请查收。'
msg.attach(MIMEText(body, 'plain', 'utf-8'))
with open(report_path, 'rb') as f:
attachment = MIMEBase('application', 'octet-stream')
attachment.set_payload(f.read())
encoders.encode_base64(attachment)
attachment.add_header('Content-Disposition', f'attachment; filename={report_path}')
msg.attach(attachment)
with smtplib.SMTP('smtp.example.com', 587) as server:
server.starttls()
server.login('[email protected]', 'your_password')
server.send_message(msg)
print(f'邮件已发送至 {to_email}')
与传统方案对比
| 维度 | 传统 Pandas 方案 | DuckDB + Python 方案 |
|---|---|---|
| 多文件读取 | 需逐个 read_csv + concat | read_csv_auto('*.csv') 一行搞定 |
| 内存占用 | 全部加载到内存 | 列式存储，按需计算 |
| 查询性能 | 百万行以上明显变慢 | 百万行秒级响应 |
| 文件支持 | 需额外库 | 原生支持 CSV/Parquet/JSON/Excel/SQLite/PostgreSQL |
| 部署复杂度 | 依赖 Pandas + 多库 | 单一 DuckDB 依赖 |
变现建议
这套自动化报表系统的价值远不止节省时间：
- SaaS 化：将报表能力封装为 API 服务，按调用次数收费，月入数千到数万元
- 数据产品：为中小企业提供行业数据报告订阅服务，年费制稳定收入
- 内部工具：在团队中推广这套系统，成为数据基础设施负责人，获得晋升机会
- 咨询培训：将实战经验转化为课程或咨询服务，单客收费 500-5000 元
核心思路：把一次性的报表工作变成可复用的数据产品。
学习更多 DuckDB 实战经验 → duckdblab.org