数据工程实战课程 · 数据存储阶段
数据仓库是面向主题、集成、非易失、随时间变化的数据集合，由Bill Inmon于1990年提出。
# 维度建模与OLAP查询（SQLite实机）
import sqlite3, time, random
from datetime import datetime, timedelta
conn = sqlite3.connect(':memory:')
conn.row_factory = sqlite3.Row
c = conn.cursor()
c.execute('CREATE TABLE dim_date (date_key INT PRIMARY KEY, full_date TEXT, year INT, quarter INT, month INT, day INT)')
c.execute('CREATE TABLE dim_product (product_key INT PRIMARY KEY, product_id TEXT, name TEXT, category TEXT, brand TEXT, price REAL)')
c.execute('CREATE TABLE dim_customer (customer_key INT PRIMARY KEY, customer_id TEXT, name TEXT, tier TEXT, region TEXT)')
c.execute('CREATE TABLE fact_sales (sale_key INTEGER PRIMARY KEY AUTOINCREMENT, date_key INT, product_key INT, customer_key INT, quantity INT, amount REAL, net_amount REAL, order_id TEXT)')
# 填充维度表
start_date = datetime(2024, 1, 1)
for i in range(365):
d = start_date + timedelta(days=i)
c.execute("INSERT INTO dim_date VALUES (?,?,?,?,?,?)", (int(d.strftime('%Y%m%d')), d.strftime('%Y-%m-%d'), d.year, (d.month-1)//3+1, d.month, d.day))
for i in range(100):
cat = random.choice(['电子','服装','食品','家居'])
c.execute("INSERT INTO dim_product VALUES (?,?,?,?,?,?)", (i+1, f"SKU_{1000+i}", f"产品_{i+1}", cat, f"品牌_{random.randint(1,10)}", round(random.uniform(10,5000),2)))
for i in range(500):
c.execute("INSERT INTO dim_customer VALUES (?,?,?,?,?)", (i+1, f"CUST_{1000+i}", f"客户_{i+1}", random.choice(['standard','premium','vip']), random.choice(['华东','华南','华北','西南','东北'])))
# 填充事实表
for i in range(5000):
qty = random.randint(1, 10)
price = round(random.uniform(10, 5000), 2)
discount = round(random.uniform(0, 0.3), 2)
net = round(price * qty * (1-discount), 2)
c.execute("INSERT INTO fact_sales (date_key,product_key,customer_key,quantity,amount,net_amount,order_id) VALUES (?,?,?,?,?,?,?)",
(random.randint(20240101,20241231), random.randint(1,100), random.randint(1,500), qty, price*qty, net, f"ORD_{random.randint(10000,99999)}"))
conn.commit()
print(f"维度: 日期365, 产品100, 客户500 | 事实: 销售5000")
print("\n--- OLAP查询 ---")
print("\n按月销售:")
for row in c.execute("SELECT d.month, SUM(f.net_amount) total, COUNT(*) orders FROM fact_sales f JOIN dim_date d ON f.date_key=d.date_key GROUP BY d.month ORDER BY d.month LIMIT 6"):
r = dict(row); print(f" {r['month']}月: ¥{r['total']:,.0f}, {r['orders']}单")
print("\n按类别:")
for row in c.execute("SELECT p.category, SUM(f.net_amount) total FROM fact_sales f JOIN dim_product p ON f.product_key=p.product_key GROUP BY p.category ORDER BY total DESC"):
r = dict(row); print(f" {r['category']}: ¥{r['total']:,.0f}")
print("\n客户分层:")
for row in c.execute("SELECT c.tier, COUNT(*) cnt, SUM(f.net_amount) total FROM fact_sales f JOIN dim_customer c ON f.customer_key=c.customer_key GROUP BY c.tier"):
r = dict(row); print(f" {r['tier']}: {r['cnt']}单, ¥{r['total']:,.0f}")
print("\n季度对比:")
for row in c.execute("SELECT d.quarter, SUM(f.net_amount) total FROM fact_sales f JOIN dim_date d ON f.date_key=d.date_key GROUP BY d.quarter"):
r = dict(row); print(f" Q{r['quarter']}: ¥{r['total']:,.0f}")
print("✅ 验证通过 - 维度建模与OLAP运行正常")🎁 下一课预告：数据湖架构！
| 策略 | 原理 | 节省比例 |
|---|---|---|
| 分层存储 | 热SSD/温HDD/冷对象存储 | 50-80% |
| 数据压缩 | 列式存储高压缩比 | 60-90% |
| 生命周期 | 自动归档删除 | 30-60% |
| 降采样 | 历史数据低精度 | 70-95% |
在生产环境中，我们经常遇到教科书不会告诉你的问题。以下是常见的实战经验和解决方案：
| 指标 | 告警阈值 | 监控方式 |
|---|---|---|
| 管道延迟 | 超过SLA 120% | 端到端时间戳追踪 |
| 数据量偏差 | 日均偏差超过30% | 与历史同期对比 |
| 错误率 | 超过0.1% | 错误计数/总记录数 |
| 数据新鲜度 | 超过SLA 150% | 最新数据时间戳 |
| 资源利用率 | CPU持续>80% | 系统监控 |
A: 根据业务延迟要求决定：秒级用流处理、分钟级用微批、小时级用批处理。大部分场景批处理足够，流处理用于实时业务场景（风控、监控、推荐）。
A: 不会完全取代，但边界在模糊。湖仓一体（如Iceberg/Delta Lake）正在融合两者优势。未来趋势是统一存储+多引擎计算。
A: 先用云原生服务（BigQuery/Snowflake + Fivetran + dbt + Looker），验证业务后再考虑自建。过早优化是万恶之源。
A: 用钱说话：数据质量问题的直接经济损失（错误决策、合规罚款、重复劳动）远大于治理投入。量化问题是第一步。
| 角色 | 职责 | 交付物 |
|---|---|---|
| 数据工程师 | 管道开发与维护 | 数据管道、基础设施 |
| 数据分析师 | 数据分析与报表 | SQL查询、BI报表 |
| 数据科学家 | 模型与实验 | ML模型、实验报告 |
| 数据产品经理 | 需求与优先级 | PRD、数据产品规划 |
| 数据管家 | 数据质量与治理 | 质量规则、数据字典 |