用 DuckDB + Streamlit 打造可售卖的 SaaS 数据分析仪表盘
💡 工具链搭建 | 难度：⭐⭐⭐ | 预计完成：1.5 小时
数据分析师最常见的困境是什么？你有一堆分析能力，但不知道如何把它变成产品卖出去。
传统做法是写 Python 脚本、部署 Flask 应用、配置 Nginx……太麻烦了。
DuckDB + Streamlit 的组合让你跳过所有这些——一个 Python 文件，一条命令，就能跑起一个交互式数据产品。而且因为 DuckDB 直接查 Parquet/CSV，你连数据库服务器都不用装。
学完本文，你能用 DuckDB + Streamlit 搭建一个可对外售卖的数据分析仪表盘产品。
一、技术栈总览
CSV/Parquet/JSON 数据源
↓
DuckDB（查询引擎）
↓
Pandas/Arrow（数据交换）
↓
Streamlit（交互界面）
↓
公网可访问的 Web 应用
这一层全部用 Python 完成，零外部依赖服务。DuckDB 直接读取本地 Parquet 文件，Streamlit 负责渲染交互式 UI。
二、第一步：创建数据层（DuckDB）
先准备数据。实际项目中这部分换成你的真实数据源。
# data_layer.py
import duckdb
import pandas as pd
import numpy as np
from pathlib import Path
class SaaSDashboard:
"""SaaS 数据分析仪表盘的核心数据层"""
def __init__(self, data_dir: str = "/tmp/saas_data"):
self.data_dir = Path(data_dir)
self.data_dir.mkdir(exist_ok=True)
self.con = duckdb.connect(":memory:")
self._load_data()
def _load_data(self):
"""加载模拟数据（替换成你的真实数据源）"""
# 客户表
np.random.seed(42)
n_customers = 500
plans = np.random.choice(
['free', 'starter', 'pro', 'enterprise'],
n_customers,
p=[0.3, 0.35, 0.25, 0.1]
)
customers = pd.DataFrame({
'customer_id': range(1, n_customers + 1),
'name': [f'客户{i}' for i in range(1, n_customers + 1)],
'plan': plans,
'signup_date': pd.date_range('2025-01-01', periods=n_customers, freq='d'),
'country': np.random.choice(
['US', 'CN', 'JP', 'DE', 'UK', 'KR', 'BR', 'IN'],
n_customers
)
})
# 使用数据
usage_records = []
base_queries = {'free': 10, 'starter': 50, 'pro': 200, 'enterprise': 1000}
for _ in range(10000):
cid = np.random.randint(1, n_customers + 1)
plan = plans[cid - 1]
usage_records.append({
'customer_id': cid,
'usage_date': (pd.Timestamp.now() - pd.Timedelta(days=np.random.randint(0, 90))).strftime('%Y-%m-%d'),
'query_count': max(0, int(np.random.normal(base_queries[plan], base_queries[plan] * 0.3))),
'storage_gb': round(np.random.uniform(0.1, 50.0), 2),
'api_calls': max(0, int(np.random.normal(base_queries[plan] * 2, base_queries[plan] * 0.5))),
'error_rate': round(max(0, np.random.normal(0.02, 0.01)), 4)
})
usage = pd.DataFrame(usage_records)
# 保存到 Parquet（生产环境这样做，查询更快）
customers.to_parquet(self.data_dir / 'customers.parquet')
usage.to_parquet(self.data_dir / 'usage.parquet')
# 注册到 DuckDB
self.con.execute("CREATE TABLE customers AS SELECT * FROM read_parquet(?)",
[str(self.data_dir / 'customers.parquet')])
self.con.execute("CREATE TABLE usage AS SELECT * FROM read_parquet(?)",
[str(self.data_dir / 'usage.parquet')])
print(f"✅ 数据加载完成：{len(customers)} 客户, {len(usage)} 条使用记录")
def get_mrr_dashboard(self) -> pd.DataFrame:
"""MRR 核心指标"""
return self.con.execute("""
SELECT
plan,
COUNT(DISTINCT customer_id) as customer_count,
SUM(CASE
WHEN plan = 'free' THEN 0
WHEN plan = 'starter' THEN 29
WHEN plan = 'pro' THEN 99
WHEN plan = 'enterprise' THEN 499
END) as total_mrr,
ROUND(
SUM(CASE
WHEN plan = 'free' THEN 0
WHEN plan = 'starter' THEN 29
WHEN plan = 'pro' THEN 99
WHEN plan = 'enterprise' THEN 499
END) * 12, 0
) as arr_projection
FROM customers
GROUP BY plan
ORDER BY total_mrr DESC
""").fetchdf()
def get_churn_risk(self, top_n: int = 20) -> pd.DataFrame:
"""流失风险客户排名"""
return self.con.execute("""
SELECT
u.customer_id,
c.name,
c.plan,
c.country,
COUNT(DISTINCT u.usage_date) as active_days,
ROUND(AVG(u.query_count), 1) as avg_daily_queries,
ROUND(AVG(u.error_rate) * 100, 2) as avg_error_rate_pct,
CASE
WHEN AVG(u.query_count) < 5 THEN '🔴 高风险'
WHEN AVG(u.query_count) < 20 THEN '🟡 中风险'
ELSE '🟢 正常'
END as risk_level
FROM usage u
JOIN customers c ON u.customer_id = c.customer_id
GROUP BY u.customer_id, c.name, c.plan, c.country
ORDER BY avg_daily_queries ASC
LIMIT ?
""", [top_n]).fetchdf()
def get_usage_trend(self, days: int = 30) -> pd.DataFrame:
"""使用趋势（按天聚合）"""
return self.con.execute("""
SELECT
usage_date,
COUNT(DISTINCT customer_id) as active_customers,
SUM(query_count) as total_queries,
ROUND(AVG(query_count), 1) as avg_queries_per_user,
SUM(api_calls) as total_api_calls,
ROUND(AVG(error_rate) * 100, 2) as avg_error_rate_pct
FROM usage
WHERE usage_date >= DATE(CURRENT_DATE - INTERVAL ? DAYS)
GROUP BY usage_date
ORDER BY usage_date
""", [days]).fetchdf()
def get_country_breakdown(self) -> pd.DataFrame:
"""国家/地区分布"""
return self.con.execute("""
SELECT
c.country,
COUNT(DISTINCT c.customer_id) as customers,
SUM(CASE
WHEN c.plan = 'starter' THEN 29
WHEN c.plan = 'pro' THEN 99
WHEN c.plan = 'enterprise' THEN 499
ELSE 0
END) as mrr,
ROUND(AVG(u.query_count), 1) as avg_usage
FROM customers c
LEFT JOIN usage u ON c.customer_id = u.customer_id
GROUP BY c.country
ORDER BY mrr DESC
""").fetchdf()
def get_customer_lifecycle(self) -> pd.DataFrame:
"""客户生命周期分析"""
return self.con.execute("""
SELECT
DATE_TRUNC('month', signup_date) as signup_month,
COUNT(*) as new_customers,
SUM(CASE WHEN plan = 'free' THEN 0
WHEN plan = 'starter' THEN 29
WHEN plan = 'pro' THEN 99
WHEN plan = 'enterprise' THEN 499
ELSE 0 END) as month_mrr,
ROUND(
SUM(CASE WHEN plan IN ('starter','pro','enterprise') THEN 1 ELSE 0 END) * 100.0 / COUNT(*),
1
) as paid_rate_pct
FROM customers
GROUP BY DATE_TRUNC('month', signup_date)
ORDER BY signup_month
""").fetchdf()
def get_revenue_forecast(self) -> pd.DataFrame:
"""收入预测（简单线性）"""
current_mrr = self.get_mrr_dashboard()['total_mrr'].sum()
forecast = pd.DataFrame({
'month': ['本月', '下月', '3个月后', '6个月后'],
'mrr': [
current_mrr,
round(current_mrr * 1.05, 0),
round(current_mrr * 1.15, 0),
round(current_mrr * 1.30, 0)
]
})
return forecast
几个关键设计点：
:memory:连接 + Parquet 文件：每次启动自动从 Parquet 加载，数据持久化在磁盘，查询结果留在内存- 参数化查询：用
?占位符传入参数，防止 SQL 注入 - 返回 Pandas DataFrame：Streamlit 直接消费 DataFrame
三、第二步：构建 Streamlit 仪表盘
# app.py
import streamlit as st
import pandas as pd
from data_layer import SaaSDashboard
st.set_page_config(page_title="SaaS 数据分析仪表盘", layout="wide")
@st.cache_data(ttl=60)
def get_dashboard():
return SaaSDashboard()
dash = get_dashboard()
st.title("📊 SaaS 数据分析仪表盘")
st.markdown("基于 DuckDB + Streamlit 构建")
# 顶部指标卡
col1, col2, col3, col4 = st.columns(4)
mrr_df = dash.get_mrr_dashboard()
total_mrr = mrr_df['total_mrr'].sum()
total_customers = mrr_df['customer_count'].sum()
paid_customers = mrr_df[mrr_df['plan'] != 'free']['customer_count'].sum()
arr = mrr_df['arr_projection'].sum()
col1.metric("总 MRR", f"${total_mrr:,}")
col2.metric("活跃客户", f"{total_customers:,}")
col3.metric("付费客户", f"{paid_customers:,}")
col4.metric("ARR 预估", f"${arr:,}")
# Tab 页签
tab1, tab2, tab3, tab4 = st.tabs(["💰 收入概览", "⚠️ 流失预警", "📈 使用趋势", "🌍 地区分布"])
# Tab1: 收入概览
with tab1:
st.subheader("各套餐 MRR 分布")
st.dataframe(mrr_df, use_container_width=True)
col_a, col_b = st.columns(2)
with col_a:
st.plotly_chart(
__import__('plotly.express').pie(mrr_df, values='total_mrr', labels='plan', title='MRR 占比'),
use_container_width=True
)
with col_b:
forecast = dash.get_revenue_forecast()
st.subheader("收入预测")
st.dataframe(forecast, use_container_width=True)
# Tab2: 流失预警
with tab2:
st.subheader("流失风险客户 TOP 20")
churn_df = dash.get_churn_risk(20)
st.dataframe(churn_df, use_container_width=True)
risk_counts = churn_df['risk_level'].value_counts()
st.bar_chart(risk_counts)
# Tab3: 使用趋势
with tab3:
days = st.slider("查看最近天数", 7, 90, 30)
trend_df = dash.get_usage_trend(days)
col_x, col_y = st.columns(2)
with col_x:
st.plotly_chart(
__import__('plotly.express').line(trend_df, x='usage_date', y='active_customers', title='活跃客户数'),
use_container_width=True
)
with col_y:
st.plotly_chart(
__import__('plotly.express').line(trend_df, x='usage_date', y='total_queries', title='总查询量'),
use_container_width=True
)
# Tab4: 地区分布
with tab4:
st.subheader("各国 MRR 与使用情况")
country_df = dash.get_country_breakdown()
st.dataframe(country_df, use_container_width=True)
st.plotly_chart(
__import__('plotly.express').bar(country_df, x='country', y='mrr', title='各国 MRR'),
use_container_width=True
)
关键技巧：
@st.cache_data：缓存查询结果 60 秒，避免每次交互都重新查询 DuckDBst.tabs：分组展示不同维度的数据st.metric：顶部展示核心 KPI 指标卡use_container_width=True：让图表和表格撑满列宽
四、与传统方案的对比
| 方案 | 数据库 | 前端框架 | 部署复杂度 | 适合场景 |
|---|---|---|---|---|
| 传统做法 | PostgreSQL | Flask/Django | ⭐⭐⭐⭐ 需配置 Nginx+Gunicorn | 大型业务系统 |
| 轻量方案 | SQLite | Flask + Charts.js | ⭐⭐⭐ 需手写 HTML/JS | 内部工具 |
| 本方案 | DuckDB | Streamlit | ⭐ 一条命令启动 | 数据产品/ MVP |
| 商业 BI | ClickHouse | Superset/Grafana | ⭐⭐⭐⭐ 资源开销大 | 企业级报表 |
DuckDB + Streamlit 的核心优势：把"我会做分析"变成"我能卖分析产品"。
五、部署到公网
用 Render 或 Railway 一键部署：
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8080
CMD ["streamlit", "run", "app.py", "--server.port=8080", "--server.headless=true"]
requirements.txt:
duckdb==1.5.4
streamlit==1.50.0
pandas==2.2.0
numpy==2.1.0
plotly==5.24.0
一条命令启动本地版本测试：
streamlit run app.py
六、变现建议
这个仪表盘产品可以直接作为以下几种商业模式的基础：
- 订阅制 SaaS：按月收费 $29/$99/$499，按套餐 tier 提供不同权限的仪表板
- 数据咨询服务：为客户定制专属分析仪表盘，按项目收费 $2000-10000
- 白标方案：将仪表盘嵌入客户的内部系统，按年收取授权费
- 数据分析即服务（DaaS）：定期生成分析报告推送给客户
核心思路：不要卖"会做分析"，要卖"能直接用的分析产品"。
七、扩展方向
- 接入真实数据源（PostgreSQL、API、Webhook）
- 添加用户认证（Streamlit + JWT）
- 集成邮件/Slack 告警（自动发送流失预警）
- 添加导出功能（PDF/Excel 报告）
学习更多 DuckDB 实战经验 → duckdblab.org