用 DuckDB 搭建多租户营收分析 SaaS，一份代码卖 10 个客户
很多自由数据分析师接到一个反复出现的需求：帮电商店主每天自动出营收分析报告。市面上常见的做法是写 Python 脚本 + 手动跑 SQL，效率低还容易出错。今天用 DuckDB 搭一个轻量级、可复用的多租户自动报告系统，你可以直接拿去改造成自己的 SaaS 产品——一份代码卖给 10 个客户，边际成本几乎为零。
为什么选 DuckDB 做 SaaS 后端
传统方案用 Pandas + CSV 处理电商订单数据，遇到千万级记录就卡住了：内存爆炸、IO 瓶颈、执行慢。DuckDB 的核心优势在于：
- 单文件即数据库：一个
.duckdb文件就是完整数据库，无需启动任何服务 - 列式存储 + 向量化执行：同样的聚合查询比 Pandas 快 10-50 倍
- 多租户天然隔离：每个店铺一张表，同一文件内数据隔离，也可以拆分为独立文件
- 零运维：不需要 PostgreSQL、MySQL 等服务端数据库
项目结构
duck_revenue_saaS/
├── config.py
├── engine.py
├── report_generator.py
├── queries/
│ ├── daily_summary.sql
│ ├── category_breakdown.sql
│ └── refund_analysis.sql
├── templates/
│ └── report.html
└── main.py
第一步：引擎层——多租户数据隔离
DuckDB 的 CREATE INDEX 支持压缩索引，比普通 B-Tree 更省空间且自动维护。
# engine.py
import duckdb
from pathlib import Path
from contextlib import contextmanager
class DuckRevenueEngine:
"""DuckDB 多租户引擎，一个文件管理所有店铺"""
def __init__(self, db_path: str = "revenue.duckdb"):
self.db_path = db_path
@contextmanager
def connection(self):
conn = duckdb.connect(self.db_path, read_only=False)
try:
yield conn
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def init_schema(self, shop_id: str):
"""为每个店铺初始化表结构"""
with self.connection() as conn:
conn.execute(f"""
CREATE TABLE IF NOT EXISTS {shop_id}_orders (
order_id VARCHAR PRIMARY KEY,
shop_id VARCHAR,
order_time TIMESTAMP,
customer_id VARCHAR,
amount DECIMAL(12, 2),
refund_amount DECIMAL(12, 2) DEFAULT 0,
status VARCHAR,
category VARCHAR,
channel VARCHAR
)
""")
conn.execute(f"""
CREATE INDEX IF NOT EXISTS idx_{shop_id}_time
ON {shop_id}_orders(order_time)
""")
conn.execute(f"""
CREATE INDEX IF NOT EXISTS idx_{shop_id}_cust
ON {shop_id}_orders(customer_id)
""")
print(f"✅ 店铺 {shop_id} 表结构初始化完成")
关键设计：所有店铺共享同一个 DuckDB 文件，通过表名前缀隔离。好处是：
- 部署极其简单——只有一个文件
- 跨店铺分析时直接 JOIN，无需分布式架构
- 备份只需拷贝一个文件
第二步：SQL 模板化查询
把查询语句拆成独立 .sql 文件，用 {{placeholder}} 做占位符，在 Python 层做替换。这样 SQL 和代码分离，方便维护和审计。
-- queries/daily_summary.sql
SELECT
shop_id,
DATE(order_time) AS order_date,
COUNT(*) AS total_orders,
SUM(amount - refund_amount) AS net_gmv,
AVG(amount - refund_amount) AS avg_order_value,
ROUND(
SUM(refund_amount) * 100.0 / NULLIF(SUM(amount), 0),
2
) AS refund_rate_pct
FROM {{shop_table}}
WHERE DATE(order_time) = DATE('{{date}}')
GROUP BY shop_id, DATE(order_time)
ORDER BY order_date DESC
-- queries/category_breakdown.sql
SELECT
category,
COUNT(*) AS order_count,
SUM(amount - refund_amount) AS category_gmv,
ROUND(
SUM(amount - refund_amount) * 100.0 /
NULLIF(SUM(SUM(amount - refund_amount)) OVER(), 0),
2
) AS gmv_share_pct
FROM {{shop_table}}
WHERE DATE(order_time) = DATE('{{date}}')
GROUP BY category
ORDER BY category_gmv DESC
LIMIT 10
-- queries/refund_analysis.sql
SELECT
DATE(order_time) AS refund_date,
order_id,
amount,
refund_amount,
customer_id,
category
FROM {{shop_table}}
WHERE DATE(order_time) = DATE('{{date}}')
AND refund_amount > 0
ORDER BY refund_amount DESC
LIMIT 20
注意 {{shop_table}} 和 {{date}} 是运行时替换的占位符，不是 SQL 参数绑定。这种模板化方式更适合批量生成报告，每个店铺独立替换。
第三步：报告生成器
# report_generator.py
import duckdb
from pathlib import Path
from datetime import date, datetime, timedelta
import jinja2
class ReportGenerator:
def __init__(self, engine, template_dir: str = "templates"):
self.engine = engine
self.env = jinja2.Environment(
loader=jinja2.FileSystemLoader(template_dir),
autoescape=True
)
def generate_report(self, shop_id: str, report_date: date) -> str:
table_name = f"{shop_id}_orders"
date_str = report_date.strftime("%Y-%m-%d")
with self.engine.connection() as conn:
# 汇总数据
summary = self._run_query(
conn, "queries/daily_summary.sql",
shop_table=table_name, date=date_str
)
# 品类分布
categories = self._run_query(
conn, "queries/category_breakdown.sql",
shop_table=table_name, date=date_str
)
# 退款明细
refunds = self._run_query(
conn, "queries/refund_analysis.sql",
shop_table=table_name, date=date_str
)
# 对比昨日
yesterday = self._run_query(
conn, "queries/daily_summary.sql",
shop_table=table_name,
date=(report_date - timedelta(days=1)).strftime("%Y-%m-%d")
)
template = self.env.get_template("report.html")
return template.render(
shop_id=shop_id,
report_date=report_date,
summary=summar.fetchone() if summary else None,
categories=categories.fetchall() if categories else [],
refunds=refunds.fetchall() if refunds else [],
yesterday=yesterday.fetchone() if yesterday else None,
now=datetime.now().strftime("%Y-%m-%d %H:%M")
)
def _run_query(self, conn, query_file: str, **params):
sql = Path(__file__).parent / query_file
sql_text = sql.read_text()
for key, value in params.items():
sql_text = sql_text.replace(f"{{{{{key}}}}}", str(value))
return conn.execute(sql_text)
第四步：主程序与调度
# main.py
import argparse
from datetime import date, timedelta
from engine import DuckRevenueEngine
from report_generator import ReportGenerator
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--date", type=str, default=date.today().isoformat())
parser.add_argument("--shops", nargs="+", required=True)
parser.add_argument("--output", default="reports/")
args = parser.parse_args()
engine = DuckRevenueEngine("revenue.duckdb")
generator = ReportGenerator(engine)
for shop_id in args.shops:
report_date = date.fromisoformat(args.date)
html = generator.generate_report(shop_id, report_date)
out_path = f"{args.output}/{shop_id}_{args.date}.html"
Path(out_path).parent.mkdir(parents=True, exist_ok=True)
Path(out_path).write_text(html)
print(f"📊 {shop_id} 报告已生成: {out_path}")
if __name__ == "__main__":
main()
用 cron 定时执行：
# 每天 22:00 生成所有店铺报告
0 22 * * * cd /home/user/duck_revenue_saaS && python main.py --shops shop001 shop002 shop003
与传统方案的性能对比
| 维度 | Pandas + CSV | DuckDB |
|---|---|---|
| 数据量上限 | 受内存限制，~500 万行后明显变慢 | 列式压缩，10 亿行轻松处理 |
| 查询速度 | 逐行迭代，聚合慢 | 向量化执行，聚合快 10-50 倍 |
| 部署复杂度 | 需要文件读写 + 内存管理 | 单文件，零配置 |
| 多租户隔离 | 需要手动拆分文件 | 表级隔离，天然支持 |
| 跨店铺分析 | 需合并多个 CSV | 直接 JOIN 同库表 |
| 内存占用 | 高（全量加载） | 低（列式 + 压缩） |
变现路径
这个系统的核心价值在于可复用性：
- 单次开发，多次售卖：一套代码适配所有店铺，只需导入不同客户的数据
- 按店铺收费：每月 200-500 元/店，10 个客户就是 2000-5000 元/月
- 增值服务：趋势分析、异常检测、 competitor 对比，每个功能都可以单独加价
- SaaS 化：把 HTML 报告升级为 Web 仪表盘，用 FastAPI 做 API 层，月收入可达 10000+
部署建议
- 轻量版：单机 Docker + cron，适合 10 个以内客户
- 标准版：FastAPI + DuckDB + Redis 缓存，支持 Web 访问
- 高级版：DuckDB Cloud 或云原生部署，支持多区域
# 快速启动
pip install duckdb jinja2
python main.py --shops shop001 shop002 --date 2026-08-18
学习更多 DuckDB 实战经验 → duckdblab.org