实验 04 · Notebook

🏬 04 · Hive 与数仓建模：四层分层 + SQL 实践

> 数仓岗的重头戏：**分层架构（ODS/DWD/DWS/ADS）** 与 **建模方法（星型/雪花/维度建模）**。
> 本节用 **sqlite3** 构建迷你数仓，手写 SQL 完成**清洗→明细→汇总→应用**四层。

📋 本节要解决的问题

| # | 问题 | 方法 |
|---|---|---|
| 1 | 数仓为什么分层？ | 解耦/复用/成本/质量 |
| 2 | 四层各做什么？ | ODS/DWD/DWS/ADS |
| 3 | 怎么建模？ | 星型/雪花/宽表 |
| 4 | 怎么用 SQL 落地？ | sqlite3 四层 ETL |

%matplotlib inline
import matplotlib
import matplotlib.pyplot as plt
matplotlib.rcParams['font.sans-serif'] = ['Microsoft YaHei', 'SimHei', 'PingFang SC', 'Noto Sans CJK SC', 'Arial Unicode MS']
matplotlib.rcParams['axes.unicode_minus'] = False
import numpy as np
import pandas as pd
import sqlite3
import os
rng = np.random.default_rng(42)

1️⃣ 数仓分层（必背框架）

前置概念：数据库 vs 数据仓库（OLTP vs OLAP）

| 维度 | 数据库（OLTP） | 数据仓库（OLAP） |
|---|---|---|
| 目标 | 高频事务：增删改单条记录 | 复杂分析：聚合扫亿级行 |
| 典型操作 | 查订单详情、扣库存 | 算 GMV、留存、转化漏斗 |
| 读写字节 | 读/写几行 | 扫全表/大范围分区 |
| 设计 | 三范式（防冗余） | 维度建模（允许冗余换速度） |
| 代表 | MySQL、PostgreSQL | Hive、Doris、ClickHouse |

事务库不适合直接跑分析（锁冲突 + 全表扫描拖垮线上业务），
所以要把数据 **T+1 同步到数仓**里做分析——这就是"分层"的起点。

print('''数仓四层：
  ODS（原始层）  ：贴源存储，原样保留（保留全量/增量，可回溯）；
  DWD（明细层）  ：清洗+标准化+维度退化，明细粒度（最细粒度事实）；
  DWS（汇总层）  ：按主题汇总（日/周维度宽表，供报表/应用）;
  ADS（应用层）  ：面向业务应用/报表/算法特征（指标结果）。

分层的价值：解耦（上游变化不传导）、复用（一致口径）、
            成本（去重存储）、质量（逐层校验）。
''')
数仓四层：
  ODS（原始层）  ：贴源存储，原样保留（保留全量/增量，可回溯）；
  DWD（明细层）  ：清洗+标准化+维度退化，明细粒度（最细粒度事实）；
  DWS（汇总层）  ：按主题汇总（日/周维度宽表，供报表/应用）;
  ADS（应用层）  ：面向业务应用/报表/算法特征（指标结果）。

分层的价值：解耦（上游变化不传导）、复用（一致口径）、
            成本（去重存储）、质量（逐层校验）。

2️⃣ 建一张"订单事实表"进 ODS（sqlite3）

if os.path.exists('mini_wh.db'):
    os.remove('mini_wh.db')
conn = sqlite3.connect('mini_wh.db')
cur = conn.cursor()
cur.execute('''CREATE TABLE ods_orders (
    order_id TEXT, user_id TEXT, item_id TEXT, city TEXT,
    amount REAL, status TEXT, ts TEXT
)''')
orders = [
    ('o1', 'u1', 'i_phone', '北京', 5999.0, 'paid', '2024-05-01 10:00:00'),
    ('o2', 'u2', 'i_phone', '上海', 5999.0, 'paid', '2024-05-01 10:05:00'),
    ('o3', 'u1', 'i_laptop', '北京', 8999.0, 'refunded', '2024-05-02 09:00:00'),
    ('o4', 'u3', 'i_tablet', '广州', 3999.0, 'paid', '2024-05-02 11:00:00'),
    ('o5', 'u2', 'i_bag', '上海', 299.0, 'paid', '2024-05-03 14:30:00'),
    ('o6', 'u4', 'i_keyboard', None, 199.0, 'paid', '2024-05-03 15:00:00'),
]
cur.executemany('INSERT INTO ods_orders VALUES (?,?,?,?,?,?,?)', orders)
conn.commit()
n = cur.execute('SELECT COUNT(*) FROM ods_orders').fetchone()[0]
print(f'ODS 层订单数: {n}')
print('➡️ ODS 几乎原样存，仅做分区/压缩；可以看到脏数据（city=None）等后续清洗')
ODS 层订单数: 6
➡️ ODS 几乎原样存，仅做分区/压缩；可以看到脏数据（city=None）等后续清洗

3️⃣ DWD：清洗 + 标准化（脏数据处理）

cur.execute('''CREATE TABLE dwd_orders AS
SELECT order_id, user_id, item_id,
       COALESCE(city, '未知') AS city,
       amount,
       CASE WHEN status = 'refunded' THEN 0 ELSE 1 END AS is_valid,
       substr(ts, 1, 10) AS dt
FROM ods_orders''')
conn.commit()
rows = cur.execute('SELECT * FROM dwd_orders').fetchall()
for r in rows:
    print(r)
print('\n➡️ DWD：补空值、状态转成可聚合标志、抽日期分区字段')
('o1', 'u1', 'i_phone', '北京', 5999.0, 1, '2024-05-01')
('o2', 'u2', 'i_phone', '上海', 5999.0, 1, '2024-05-01')
('o3', 'u1', 'i_laptop', '北京', 8999.0, 0, '2024-05-02')
('o4', 'u3', 'i_tablet', '广州', 3999.0, 1, '2024-05-02')
('o5', 'u2', 'i_bag', '上海', 299.0, 1, '2024-05-03')
('o6', 'u4', 'i_keyboard', '未知', 199.0, 1, '2024-05-03')

➡️ DWD：补空值、状态转成可聚合标志、抽日期分区字段

4️⃣ DWS：按主题日汇总（宽表雏形）

cur.execute('''CREATE TABLE dws_daily_orders AS
SELECT dt, city,
       COUNT(*) AS order_cnt,
       COUNT(DISTINCT user_id) AS user_cnt,
       SUM(amount) AS gmv,
       SUM(is_valid) AS valid_cnt
FROM dwd_orders
GROUP BY dt, city''')
conn.commit()
rows = cur.execute('SELECT * FROM dws_daily_orders ORDER BY dt, city').fetchall()
print('DWS 日汇总（dt, city, 订单数, 用户数, GMV, 有效单）:')
for r in rows:
    print(' ', r)
print('➡️ DWS 就是"主题宽表"：一次计算多指标，供多处复用')
DWS 日汇总（dt, city, 订单数, 用户数, GMV, 有效单）:
  ('2024-05-01', '上海', 1, 1, 5999.0, 1)
  ('2024-05-01', '北京', 1, 1, 5999.0, 1)
  ('2024-05-02', '北京', 1, 1, 8999.0, 0)
  ('2024-05-02', '广州', 1, 1, 3999.0, 1)
  ('2024-05-03', '上海', 1, 1, 299.0, 1)
  ('2024-05-03', '未知', 1, 1, 199.0, 1)
➡️ DWS 就是"主题宽表"：一次计算多指标，供多处复用

5️⃣ ADS：应用层指标（Top 城市 / 全量汇总）

top_city = cur.execute('''
    SELECT city, SUM(gmv) AS gmv FROM dws_daily_orders
    GROUP BY city ORDER BY gmv DESC LIMIT 3''').fetchall()
print('GMV Top3 城市:')
for c, g in top_city:
    print(f'  {c}: {g:.0f} 元')
total = cur.execute('''
    SELECT SUM(order_cnt), SUM(user_cnt), SUM(gmv), SUM(valid_cnt)
    FROM dws_daily_orders''').fetchone()
distinct_users = cur.execute('SELECT COUNT(DISTINCT user_id) FROM dwd_orders').fetchone()[0]
print(f'全量: 订单 {total[0]}, 用户(SUM) {total[1]}, 用户(去重) {distinct_users}, '
      f'GMV {total[2]:.0f} 元, 有效单 {total[3]}')
print('➡️ 教学点：DWS 按(天,城市)分组 COUNT(DISTINCT user_id) 后 SUM 会重复计数，'
      '跨组去重需单独统计（如按天去重）')
assert total[1] >= 4 and distinct_users == 4
print('✅ ADS 指标正确（去重用户数=4：u1,u2,u3,u4）')
GMV Top3 城市:
  北京: 14998 元
  上海: 6298 元
  广州: 3999 元
全量: 订单 6, 用户(SUM) 6, 用户(去重) 4, GMV 25494 元, 有效单 5
➡️ 教学点：DWS 按(天,城市)分组 COUNT(DISTINCT user_id) 后 SUM 会重复计数，跨组去重需单独统计（如按天去重）
✅ ADS 指标正确（去重用户数=4：u1,u2,u3,u4）

# 可视化：四层各层的行数与粒度（自下而上加工）
import matplotlib.pyplot as plt
fig, ax = plt.subplots(figsize=(9, 4.2))
layers = ['ODS(原始订单)', 'DWD(清洗明细)', 'DWS(日×城市汇总)', 'ADS(Top城市)']
# 实际行数来自上面建的表
cnt = [
    cur.execute('SELECT COUNT(*) FROM ods_orders').fetchone()[0],
    cur.execute('SELECT COUNT(*) FROM dwd_orders').fetchone()[0],
    cur.execute('SELECT COUNT(*) FROM dws_daily_orders').fetchone()[0],
    len(top_city),
]
bars = ax.bar(layers, cnt, color=['#D9E2F3', '#E2EFDA', '#FBE5D6', '#F8CBAD'],
              ec=['#2F5597', '#548235', '#ED7D31', '#C00000'], width=0.55)
for b, v in zip(bars, cnt):
    ax.text(b.get_x() + b.get_width() / 2, v + 0.15, str(v), ha='center', fontsize=12, weight='bold')
ax.set_ylabel('行数'); ax.set_title('四层粒度变化：明细行 → 汇总行 → 指标（数据越上层越小、越直接可用）')
plt.show()
Notebook 已保存的运行结果

6️⃣ 建模方法：星型 / 雪花 / 宽表

print('''事实表 + 维度表：
  星型模型：事实表居中，维度表直接连（冗余维度，查询快）；
  雪花模型：维度表再规范化拆分（省空间，查询多 join）；
  宽表（维度退化）：把常用维度字段直接冗余进事实表（DWS 常用，最快）。

选型口诀：查询性能优先 -> 星型/宽表；存储与一致性优先 -> 雪花。
数仓工具：Hive（离线）、Presto/StarRocks/Doris（即席/实时）、Iceberg/Hudi（湖仓）。
''')
事实表 + 维度表：
  星型模型：事实表居中，维度表直接连（冗余维度，查询快）；
  雪花模型：维度表再规范化拆分（省空间，查询多 join）；
  宽表（维度退化）：把常用维度字段直接冗余进事实表（DWS 常用，最快）。

选型口诀：查询性能优先 -> 星型/宽表；存储与一致性优先 -> 雪花。
数仓工具：Hive（离线）、Presto/StarRocks/Doris（即席/实时）、Iceberg/Hudi（湖仓）。

7️⃣ 面试高频：数仓治理

缓慢变化维（SCD）与拉链表

维度属性会变化（用户搬家了城市从北京→上海）。两种处理：

• **Type 1（覆盖）**：直接 UPDATE 新值。简单，但**丢失历史**；

• **Type 2（版本）**：旧行改 `end_date` 封版，插新行 `start_date` 生效，`end_date=9999-12-31` 表示当前有效。

可回答"3 月份这个用户在哪个城市"——历史可溯。

**拉链表**就是 Type 2 的落地形式：`user_id, 属性, start_date, end_date` 四列，
每天用"昨天的闭链 + 今天的新开链"两条 SQL 完成更新。适合**变化缓慢、量大**的维表（用户/商品）。

下面用 sqlite3 演示一次完整的拉链更新。

# 拉链表演示：u1 的 city 从 北京 -> 上海（2024-04-01 生效）
cur.execute('CREATE TABLE IF NOT EXISTS d_user_zip (user_id TEXT, city TEXT, start_date TEXT, end_date TEXT)')
cur.execute('DELETE FROM d_user_zip')
cur.executemany('INSERT INTO d_user_zip VALUES (?,?,?,?)', [
    ('u1', '北京', '2024-01-01', '9999-12-31'),
    ('u2', '上海', '2024-01-01', '9999-12-31'),
])
conn.commit()

def zipper_update(user, new_city, eff_date):
    # 1) 闭链：当前有效记录的 end_date 改为 生效日前一天
    cur.execute("UPDATE d_user_zip SET end_date = ? WHERE user_id=? AND end_date='9999-12-31'",
                (eff_date, user))
    # 2) 开链：插入新版本
    cur.execute('INSERT INTO d_user_zip VALUES (?,?,?,?)', (user, new_city, eff_date, '9999-12-31'))
    conn.commit()

zipper_update('u1', '上海', '2024-04-01')
rows = cur.execute('SELECT * FROM d_user_zip ORDER BY user_id, start_date').fetchall()
print('拉链表（user, city, 起始日, 截止日）:')
for r in rows:
    print('  ', r)
# 验证：2024-03-15 查 u1 应在北京，2024-05-01 查 u1 应在上海
q = lambda d: cur.execute(
    "SELECT city FROM d_user_zip WHERE user_id='u1' AND ? >= start_date AND ? <= end_date", (d, d)
).fetchone()[0]
assert q('2024-03-15') == '北京' and q('2024-05-01') == '上海'
print('\n✅ 按时间点查询历史正确：3月15日在北京，5月1日在上海')
拉链表（user, city, 起始日, 截止日）:
   ('u1', '北京', '2024-01-01', '2024-04-01')
   ('u1', '上海', '2024-04-01', '9999-12-31')
   ('u2', '上海', '2024-01-01', '9999-12-31')

✅ 按时间点查询历史正确：3月15日在北京，5月1日在上海

Hive 本身是什么（面试常混）

Hive 不是数据库，而是**把 SQL 翻译成分布式任务的引擎**：

• 表结构（schema）存在 **Metastore**（元数据服务，常配 MySQL）；

• 数据存在 HDFS/对象存储上（就是普通文件）；

• 查询时 HiveSQL → 解析器 → 优化器 → 执行计划（早期 MR，现 Tez/Spark 引擎）；

• **分区（partition）**＝按目录切分（如 `dt=2024-05-01/`），查询带分区条件可跳过无关目录（裁剪）；

• **分桶（bucket）**＝按 key 哈希切文件，利于采样与 map-side join。

print('''1) 指标口径：同一指标全局唯一（命名/计算/来源统一）；
2) 数据质量：完整性/准确性/一致性/及时性 四大维度 + 质量监控；
3) 元数据管理：表血缘、字段字典、变更管理；
4) 权限与安全：行列级权限、脱敏、审计；
5) 数据倾斜：大 key 打散/两阶段聚合/倾斜检测；
6) 缓慢变化维（SCD）：维度属性变化如何处理（Type1 覆盖/Type2 加版本）。
''')
1) 指标口径：同一指标全局唯一（命名/计算/来源统一）；
2) 数据质量：完整性/准确性/一致性/及时性 四大维度 + 质量监控；
3) 元数据管理：表血缘、字段字典、变更管理；
4) 权限与安全：行列级权限、脱敏、审计；
5) 数据倾斜：大 key 打散/两阶段聚合/倾斜检测；
6) 缓慢变化维（SCD）：维度属性变化如何处理（Type1 覆盖/Type2 加版本）。

📝 自测清单

• [ ] 能画出 ODS/DWD/DWS/ADS 四层与职责

• [ ] 能用 SQL 完成一层 ETL（清洗/聚合）

• [ ] 能对比星型/雪花/宽表

• [ ] 能答 6 个数仓治理高频点

• [ ] 能解释指标口径统一与数据质量四维度