DuckDB + SQLite 零成本搭建轻量级 BI 仪表盘

pip install duckdb fastapi uvicorn pandas pyarrow

import sqlite3
import random
from datetime import datetime, timedelta

conn = sqlite3.connect("business.db")
cur = conn.cursor()

cur.execute("""
CREATE TABLE IF NOT EXISTS orders (
    id INTEGER PRIMARY KEY,
    order_date TEXT,
    customer_id TEXT,
    product TEXT,
    quantity INTEGER,
    amount REAL,
    region TEXT
)
""")

cur.execute("""
CREATE TABLE IF NOT EXISTS customers (
    id TEXT PRIMARY KEY,
    name TEXT,
    signup_date TEXT,
    tier TEXT
)
""")

# 生成 500 条订单
random.seed(42)
products = ["Pro版", "企业版", "基础版", "API调用", "存储扩容"]
regions = ["华东", "华南", "华北", "西南", "华中"]

for i in range(1, 501):
    date = datetime(2025, 1, 1) + timedelta(days=random.randint(0, 365))
    cur.execute(
        "INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?, ?)",
        (i, date.strftime("%Y-%m-%d"), f"C{random.randint(1,100):03d}",
         random.choice(products), random.randint(1, 20),
         round(random.uniform(50, 5000), 2), random.choice(regions))
    )

# 生成 100 个客户
customers = [
    (f"C{i:03d}", f"客户{i}", f"2024-{random.randint(1,12):02d}-01", random.choice(["免费", "标准", "高级"]))
    for i in range(1, 101)
]
cur.executemany("INSERT INTO customers VALUES (?, ?, ?, ?)", customers)

conn.commit()
conn.close()
print(f"✅ 已创建 {len(customers)} 个客户、500 条订单")

import duckdb

con = duckdb.connect()

# 关键：ATTACH SQLite 文件
con.execute("ATTACH 'business.db' AS biz (TYPE SQLITE)")

# 查询 1：月度营收趋势
monthly_revenue = con.execute("""
    SELECT
        DATE_TRUNC('month', order_date::DATE) AS month,
        SUM(amount) AS revenue,
        COUNT(*) AS orders,
        ROUND(AVG(amount), 2) AS avg_order_value
    FROM biz.orders
    GROUP BY month
    ORDER BY month
""").fetchdf()

print("📈 月度营收趋势:")
for _, row in monthly_revenue.iterrows():
    print(f"  {row.month}: ¥{row.revenue:,.0f} | 订单 {int(row.orders)} | 客单价 ¥{row.avg_order_value:.0f}")

# 查询 2：客户分层 + 消费能力交叉分析
customer_analysis = con.execute("""
    WITH customer_orders AS (
        SELECT
            c.id, c.tier,
            SUM(o.amount) AS total_spend,
            COUNT(DISTINCT o.product) AS products_bought,
            COUNT(*) AS order_count
        FROM biz.customers c
        LEFT JOIN biz.orders o ON c.id = o.customer_id
        GROUP BY c.id, c.tier
    )
    SELECT
        tier,
        COUNT(*) AS customer_count,
        ROUND(AVG(total_spend), 0) AS avg_spend,
        ROUND(SUM(total_spend), 0) AS total_revenue,
        ROUND(AVG(order_count), 1) AS avg_orders
    FROM customer_orders
    GROUP BY tier
    ORDER BY total_revenue DESC
""").fetchdf()

print("\n👥 客户分层分析:")
for _, row in customer_analysis.iterrows():
    print(f"  {row.tier}: {int(row.customer_count)} 人 | 人均消费 ¥{row.avg_spend:,.0f} | 总营收 ¥{row.total_revenue:,.0f}")

# 查询 3：找出高价值客户
vip_list = con.execute("""
    SELECT c.id, c.name, c.tier, SUM(o.amount) AS total_spend, COUNT(*) AS order_count
    FROM biz.customers c
    JOIN biz.orders o ON c.id = o.customer_id
    GROUP BY c.id, c.name, c.tier
    HAVING SUM(o.amount) > 5000 OR COUNT(*) >= 10
    ORDER BY total_spend DESC
    LIMIT 20
""").fetchall()

print("\n⭐ VIP 客户 TOP 20:")
for v in vip_list:
    print(f"  {v[1]} ({v[0]}): ¥{v[3]:,.0f} | {int(v[4])} 单")

import json

report = {
    "monthly_revenue": monthly_revenue.to_dict(orient="records"),
    "tier_analysis": customer_analysis.to_dict(orient="records"),
    "vip_customers": [list(v) for v in vip_list]
}

with open("analysis_report.json", "w", encoding="utf-8") as f:
    json.dump(report, f, ensure_ascii=False, indent=2)

print("✅ 报告已导出到 analysis_report.json")

from fastapi import FastAPI
import duckdb

app = FastAPI()

@app.get("/api/revenue/monthly")
def get_monthly_revenue():
    con = duckdb.connect()
    con.execute("ATTACH 'business.db' AS biz (TYPE SQLITE)")
    result = con.execute("""
        SELECT DATE_TRUNC('month', order_date::DATE) AS month,
               SUM(amount) AS revenue, COUNT(*) AS orders
        FROM biz.orders GROUP BY month ORDER BY month
    """).fetchdf().to_dict(orient="records")
    con.close()
    return result

@app.get("/api/customers/vip")
def get_vip_customers():
    con = duckdb.connect()
    con.execute("ATTACH 'business.db' AS biz (TYPE SQLITE)")
    result = con.execute("""
        SELECT c.id, c.name, SUM(o.amount) AS total_spend, COUNT(*) AS order_count
        FROM biz.customers c JOIN biz.orders o ON c.id = o.customer_id
        GROUP BY c.id, c.name
        HAVING SUM(o.amount) > 5000
        ORDER BY total_spend DESC LIMIT 50
    """).fetchall()
    con.close()
    return [{"id": r[0], "name": r[1], "total_spend": r[2], "order_count": r[3]} for r in result]

uvicorn main:app --host 0.0.0.0 --port 8000

import duckdb

con = duckdb.connect()
con.execute("ATTACH 'business.db' AS biz (TYPE SQLITE)")

# 导出为 Parquet
con.execute("COPY (SELECT * FROM biz.orders) TO 'orders.parquet' (FORMAT PARQUET)")

# 之后直接用 Parquet 查询
parquet_con = duckdb.connect()
result = parquet_con.execute("""
    SELECT DATE_TRUNC('month', order_date::DATE) AS month,
           SUM(amount) AS revenue
    FROM read_parquet('orders.parquet')
    GROUP BY month ORDER BY month
""").fetchdf()