60.1 数据平台分层
一句话
数据平台的分层不是把组件摆成五排，而是把「谁生产数据、谁消费数据、谁担保 SLA、谁付存储账单」切成四份可执行的契约；层的边界由元数据 + 依赖规则定义，产品名只是实现。
先破除五个误区
| 误区 | 现实 |
|---|---|
| 平台 = 组件堆叠（Kafka + Spark + 数仓） | 组件是能力，层是责任边界；没有 owner 和 SLA 的组件集合叫「一堆中间件」 |
| 分层越多越专业 | 每多一层就多一次延迟、一份存储、一个故障域；层的价值是复用与隔离，不是数量 |
| 上云 = 有数据平台 | 托管服务只替你运维组件，不替你定义口径、权限与质量门禁 |
| 数据湖 = 数据平台 | 湖只是存储形态；没有 catalog、表格式与治理，湖会退化成「更贵的网盘」 |
| 建平台 = 建中台 | 中台是组织与复用问题；平台是技术契约问题，两者不能互相替代 |
🔑 判断一个「层」是否成立，只问三件事：它有没有独立的输入输出？有没有独立的 SLA？有没有明确的 owner？ 三问缺一，这一层就是架构图上的装饰。
五层：每层只回答一个问题
| 层 | 回答什么 | 输入 → 输出 | 关键指标 | 典型产品 | 典型失败 |
|---|---|---|---|---|---|
| 接入 | 数据怎么进来 | 业务库 / 日志 / API / 文件 → 原始数据 | 端到端延迟、丢数率、重复率 | Kafka、Debezium、Flink CDC、Airbyte、DataX、Filebeat | 漏采、重复投递、schema 漂移 |
| 存储 | 放哪、怎么组织 | 原始 / 明细 / 汇总 → 可查询数据集 | 单位 GB 成本、持久性、可用性 | S3 / OSS、HDFS、Iceberg / Delta / Hudi、ClickHouse | 小文件、元数据瓶颈、冷热错配 |
| 计算 | 怎么算 | 数据集 → 数据集 | 任务成功率、SLA 达成率、单位计算成本 | Spark、Flink、Trino、DuckDB、dbt | 数据倾斜、失败重跑、资源争抢 |
| 服务 | 怎么给出去 | 数据集 → API / 报表 / 特征 | P99、并发、口径一致率 | Trino / Presto、ClickHouse、BI、特征平台 | 慢查询拖垮集群、同名指标两个值 |
| 治理 | 谁负责、能不能信 | 元数据 / 规则 → 策略与账本 | 覆盖率、违规数、成本归因率 | DataHub、OpenMetadata、Great Expectations、Ranger | 元数据过期、规则形同虚设 |
这张表最重要的不是产品名，而是每层的失败模式不同：接入层怕丢和重，存储层怕小文件和成本，计算层怕倾斜和排队，服务层怕慢查询和口径，治理层怕「有规则、没执行」。用一套告警去覆盖五层，等于没有告警。
控制面与数据面：必须分开
| 控制面 | 数据面 | |
|---|---|---|
| 组成 | 调度、元数据、权限、质量规则、成本配额 | 存储、计算、网络、查询执行 |
| 流量特征 | 低频写、高频读、强一致 | 高频读写、可容忍短暂不一致 |
| 故障影响 | 不能调度新任务 / 新策略不生效，已跑任务可继续 | 业务直接查不到数 |
| 变更方式 | 审计、灰度、可回滚 | 限流、降级、隔离 |
| 部署要求 | 多副本 + 客户端缓存 + 变更通知 | 资源池隔离 + 熔断 |
把元数据库和查询集群混在一套实例里，最常见的结局是：一次元数据全量扫描把查询集群的 IO 打满，所有看板同时超时。控制面故障应该只影响「变更」，不该影响「已经能跑的东西」。
🔑 控制面是变更的来源，必须可审计、可灰度、可回滚；数据面是流量的通道，必须可限流、可降级、可隔离。两者的 SLO 要分开定，事故也要分开复盘。
一条订单数据的旅程：每层的时间预算
| 跳 | 层 | 动作 | 示例预算（P99） | 失败后多久可见 |
|---|---|---|---|---|
| 1 | 接入 | 业务库 CDC → ODS | 5 min | 延迟告警 10 min 内 |
| 2 | 计算 | ODS → DWD（清洗、去重、维表关联） | 15 min | 任务失败告警 |
| 3 | 计算 | DWD → DWS（按主题汇总） | 30 min | SLA 看板变红 |
| 4 | 计算 | DWS → ADS（指标口径） | 30 min | 指标对账差异 |
| 5 | 服务 | ADS → BI / API | 5 min | 页面/接口错误率 |
把这条链的预算加起来是 85 分钟——它不是「技术上限」，而是从业务问题倒推出来的：运营早上 9 点要看昨日 GMV，T+1 批处理完全够；大促要分钟级，就得把 2–4 跳改成流式（见 62.6 实时数仓）。
⚠️ 延迟预算必须按 P99 + 排队时间 定，不能按平均耗时。晚高峰的排队时间往往比计算时间长一个数量级，而平均值会把它抹平。
分层契约：每个数据集必须登记的字段
| 字段 | 缺了会怎样 | 谁维护 |
|---|---|---|
owner | 口径错了没人认领，告警没人接 | 数据开发团队 |
layer | 依赖方向无法校验，越层无人发现 | 平台（从命名/注册信息生成） |
sla（新鲜度 + 产出时间） | 无法判断「晚到」是故障还是正常 | 数据开发 + 业务方 |
classification（L1–L4） | 权限与脱敏无法执行（见 38.5 数据分级与脱敏） | 安全 + 数据 owner |
retention | 成本无限增长，合规删除无依据（见 46.6 数据生命周期） | 数据 owner + 合规 |
upstream / consumers | 影响面算不出来，改表靠「谁记得」 | 平台自动采集 |
🔑 分层的可执行形态是「数据集清单 + 依赖规则 + CI 检查」，不是一张架构图。图会过期，CI 不会。
代码：把分层规则做成 CI 门禁
下面这段程序把四条规则变成可执行检查：R1 依赖只能从上游指向下游；R2 ODS 不能跳过 DWD 直写 DWS/ADS；R3 数据集四项元数据必须齐全；R4 APP 是终端，不能被当作输入。
#!/usr/bin/env python3
from dataclasses import dataclass
RANK = {"ODS": 0, "DWD": 1, "DWS": 2, "ADS": 3, "APP": 4}
@dataclass
class Dataset:
name: str
layer: str
owner: str = ""
sla_minutes: int = 0
classification: str = ""
retention_days: int = 0
exception: str = "" # 越层审批工单号
@dataclass
class Pipeline:
name: str
inputs: list[str]
output: str
owner: str = ""
DATASETS = {d.name: d for d in [
Dataset("ods_orders", "ODS", "data-eng", 5, "L3", 365),
Dataset("ods_users", "ODS", "data-eng", 30, "L3", 365),
Dataset("ods_payments", "ODS", "pay-eng", 0, "L4", 0), # 缺 SLA + 留存
Dataset("dwd_trade_order_di", "DWD", "dw-trade", 15, "L3", 1095),
Dataset("dwd_user_profile", "DWD", "dw-user", 60, "L3", 1095),
Dataset("dws_trade_sales_1d", "DWS", "dw-trade", 30, "L2", 1095),
Dataset("dws_user_value_1d", "DWS", "", 60, "L3", 730), # 缺 owner
Dataset("ads_gmv_dashboard", "ADS", "bi-team", 30, "L2", 365),
Dataset("ads_finance_report", "ADS", "fin-team", 120, "L3", 1825),
Dataset("ml_churn_feature", "APP", "ml-team", 60, "L3", 365),
]}
PIPELINES = [
Pipeline("sync_orders", ["ods_orders"], "dwd_trade_order_di", "dw-trade"),
Pipeline("sync_users", ["ods_users"], "dwd_user_profile", "dw-user"),
Pipeline("sales_daily", ["dwd_trade_order_di"], "dws_trade_sales_1d", "dw-trade"),
Pipeline("user_value", ["dwd_user_profile", "dws_trade_sales_1d"], "dws_user_value_1d", "dw-user"),
Pipeline("gmv_dashboard", ["dws_trade_sales_1d"], "ads_gmv_dashboard", "bi-team"),
Pipeline("finance_report", ["ods_payments"], "ads_finance_report", "fin-team"), # 越层
Pipeline("churn_feature", ["ads_finance_report"], "ml_churn_feature", "ml-team"), # 合法
Pipeline("ops_backfill", ["ads_gmv_dashboard"], "dwd_user_profile"), # 反向
Pipeline("share_feature", ["ml_churn_feature"], "ads_finance_report", "bi-team"), # APP 被消费
]
def check():
errs, warns = [], []
for p in PIPELINES:
if not p.owner:
warns.append(f"[W] 任务 {p.name} 没有 owner")
out = DATASETS.get(p.output)
if out is None:
errs.append(f"[R0] {p.name} 的输出数据集 {p.output} 未登记")
continue
for src in p.inputs:
s = DATASETS.get(src)
if s is None:
errs.append(f"[R0] {p.name} 的输入数据集 {src} 未登记")
continue
if RANK[s.layer] > RANK[out.layer]:
errs.append(f"[R1] {p.name}: {src}({s.layer}) -> {p.output}({out.layer}) 反向依赖")
if s.layer == "ODS" and out.layer in ("DWS", "ADS") and not out.exception:
errs.append(f"[R2] {p.name}: ODS 跳过 DWD 直写 {out.layer}（无例外工单）")
if s.layer == "APP":
errs.append(f"[R4] {p.name}: APP 数据集 {src} 被当作输入")
for d in DATASETS.values():
missing = []
if not d.owner:
missing.append("owner")
if d.sla_minutes <= 0:
missing.append("SLA")
if not d.classification:
missing.append("分级")
if d.retention_days <= 0:
missing.append("留存期")
if missing:
errs.append(f"[R3] {d.name}({d.layer}) 缺：{', '.join(missing)}")
return errs, warns
def main():
errs, warns = check()
checks = len(DATASETS) * 4 + len(PIPELINES)
score = max(0, 100 - 4 * len(errs) - 2 * len(warns))
print(f"数据集 {len(DATASETS)} 个，任务 {len(PIPELINES)} 个，检查项 {checks} 条")
print(f"硬错误 {len(errs)} 条，告警 {len(warns)} 条，平台分层健康分 {score}\n")
for e in errs:
print(" ✗", e)
for w in warns:
print(" ⚠", w)
layers = ["ODS", "DWD", "DWS", "ADS", "APP"]
counts = {(a, b): 0 for a in layers for b in layers}
for p in PIPELINES:
for src in p.inputs:
if src in DATASETS and p.output in DATASETS:
counts[(DATASETS[src].layer, DATASETS[p.output].layer)] += 1
print("\n分层依赖矩阵（行=输入层，列=输出层）：")
print(" " + "".join(f"{x:>6}" for x in layers))
for a in layers:
print(f" {a:<4}" + "".join(f"{counts[(a, b)]:>6}" for b in layers))
print("\n按层统计：")
for layer in layers:
ds = [d for d in DATASETS.values() if d.layer == layer]
ready = sum(1 for d in ds if d.owner and d.sla_minutes and d.classification and d.retention_days)
print(f" {layer:<4} 数据集 {len(ds)} 个，元数据完整 {ready}/{len(ds)}")
main()
本机实测（macOS，Apple M1，Python 3.14.8）：
数据集 10 个，任务 9 个，检查项 49 条
硬错误 6 条，告警 1 条，平台分层健康分 74
✗ [R2] finance_report: ODS 跳过 DWD 直写 ADS（无例外工单）
✗ [R1] ops_backfill: ads_gmv_dashboard(ADS) -> dwd_user_profile(DWD) 反向依赖
✗ [R1] share_feature: ml_churn_feature(APP) -> ads_finance_report(ADS) 反向依赖
✗ [R4] share_feature: APP 数据集 ml_churn_feature 被当作输入
✗ [R3] ods_payments(ODS) 缺：SLA, 留存期
✗ [R3] dws_user_value_1d(DWS) 缺：owner
⚠ [W] 任务 ops_backfill 没有 owner
分层依赖矩阵（行=输入层，列=输出层）：
ODS DWD DWS ADS APP
ODS 0 2 0 1 0
DWD 0 0 2 0 0
DWS 0 0 1 1 0
ADS 0 1 0 0 1
APP 0 0 0 1 0
按层统计：
ODS 数据集 3 个，元数据完整 2/3
DWD 数据集 2 个，元数据完整 2/2
DWS 数据集 2 个，元数据完整 1/2
ADS 数据集 2 个，元数据完整 2/2
APP 数据集 1 个，元数据完整 1/1
读数：
| 信号 | 说明 | 修法 |
|---|---|---|
ODS → ADS 一条 | 支付报表绕开 DWD，等于把清洗逻辑复制到报表里 | 补 DWD 支付明细表；申请例外必须有工单 + 到期日 |
| 反向依赖两条 | ADS 被回灌成 DWD 的输入、APP 被当作输入 | 改依赖，或把这些任务降级为「临时任务」不进生产 DAG |
| 元数据缺口三处 | 缺 owner / SLA / 留存，故障时无人认领 | 注册时强校验；存量按周清理 |
| 矩阵对角线 | DWS→DWS 有一条（用户价值汇总） | 同层依赖允许，但要限制层数，避免 DAG 变成蜘蛛网 |
⚠️ 门禁最容易死在「存量全红」。上线分三步：① 只报告不阻断，跑两周把存量清单交给 owner；② 只阻断本次变更新增的违规（存量加豁免期）；③ 豁免到期前清理，再全量阻断。一步到位只会逼团队加白名单，最后规则失效。
平台分层怎么落到组织
| 角色 | 负责 | 不负责 | 判据 |
|---|---|---|---|
| 平台团队 | 存储、计算、调度、catalog、权限基础设施 | 指标口径、业务模型 | 平台 SLO（调度成功率、查询可用性） |
| 数据开发 | DWD / DWS 模型、SLA、数据质量 | 业务指标定义 | 数据集 owner 与 on-call |
| 分析 / BI | ADS、报表、指标口径 | 底层链路运维 | 指标字典签字 |
| 安全 / 合规 | 分级、授权、审计、删除 | 业务建模 | 审计通过率 |
| 业务方 | 指标定义、验收、优先级 | 技术实现 | 口径变更发起人 |
🔑 谁 on-call，谁拥有 SLA；谁定义指标，谁签字口径。两者不能都推给「平台团队」。
选型：三种典型形态
| 形态 | 适用 | 组件组合 | 什么时候别用 |
|---|---|---|---|
| 托管 ELT | < 10 名数据工程师、< 10 TB、无强实时 | Fivetran / Airbyte + BigQuery / Snowflake + dbt + Looker / Superset | 数据不能出域、成本按扫描量不可控 |
| 自建湖仓 | 数据量大、要多引擎、要开放格式 | Kafka + 对象存储 + Iceberg + Spark / Flink + Trino + Airflow + DataHub + GE + Ranger | 团队 < 3 人：控制面运维成本会吃掉全部收益 |
| 实时优先 | 风控、推荐、实时大屏 | Kafka + Flink + ClickHouse / Doris / StarRocks + 轻量调度 | 业务只要 T+1：流式带来的状态与运维复杂度不划算 |
💡 我的判断：先买「控制面」，再自建「数据面」。元数据、调度、权限这些控制面组件自研的收益极低、维护成本极高；而存储与计算直接决定成本与性能，值得按业务自建或深度调优。
生产落地：指名 → 可抄 → 可验 → 可退
指名：元数据用 DataHub（替代 OpenMetadata / Atlas）；质量用 Great Expectations（替代 Soda、dbt tests 视栈而定）；湖表格式用 Apache Iceberg（替代 Delta Lake / Hudi，按引擎生态选）；权限用 Apache Ranger（云上用 Lake Formation / Unity Catalog）。
可抄（未验证：本机未安装 Docker / DataHub，版本 0.15.x，GMS 默认 8080、前端 9002）：
# 1) 启动单机 DataHub（需要 Docker，建议 ≥ 8 GB 内存）
python3 -m pip install --upgrade 'acryl-datahub'
datahub docker quickstart
# 2) 采集 PostgreSQL 元数据（recipe.yml 见下）
datahub ingest -c recipe.yml
# recipe.yml：只读账号；不要用业务写账号
source:
type: postgres
config:
host: localhost
port: 5432
username: meta_reader
password: ${PG_PASSWORD}
database: shop
include_tables: true
sink:
type: datahub-rest
config:
server: "http://localhost:8080"
可验：本节脚本可直接跑，输出 6 条硬错误即为门禁生效；产品侧验证用 curl -s localhost:8080/health（预期 {"status":"UP"}，未验证）与前端搜索数据集名。可退：采集器是只读旁路，停掉不影响业务；门禁先以 --warn-only 运行，确认误报率后切阻断；权限策略用 shadow 模式（见 49.3 权限中心与权限矩阵）。
反模式
| 反模式 | 后果 |
|---|---|
| 按产品名分层（「Kafka 层 / Spark 层」） | 换一个组件就要重画架构，层失去稳定含义 |
| 平台团队同时管基础设施与指标口径 | 出事互相甩锅，口径变更无人签字 |
| 所有数据先入湖、不登记 owner / SLA | 三个月后变成数据沼泽，谁也不敢删 |
| ADS 直读 ODS | 清洗逻辑在报表里复制 N 份，上游改字段全站爆炸 |
| 控制面与数据面共用一套集群 | 元数据扫描 / 权限服务重启拖垮全部查询 |
| 用架构图代替依赖规则 | 图在评审时正确，三个月后与真实链路脱节 |
| 只建平台不建成本账 | 存储与计算成本随数据量线性涨，第二年预算翻倍 |
| 例外没有到期日 | 「临时越层」永久化，门禁形同虚设 |
自检
Q1 判断一个「层」是否成立，用哪三个问题？为什么产品名不能作为判据？
▸ 答案：三问是「有没有独立输入输出、有没有独立 SLA、有没有明确 owner」。产品名不能作判据，因为同一层可以用不同产品实现（对象存储可以是 S3、OSS、HDFS），而同一产品也可能横跨多层（Spark 既能做接入后的清洗，也能做服务层查询）。层的本质是责任与契约的切分，产品只是实现手段；按产品分层，换组件就会导致架构返工。
Q2 控制面与数据面分开，具体防住哪类故障？举一个反例。
▸ 答案：防的是「控制面故障放大成数据面全站不可用」。例如元数据库与查询集群共用实例，一次元数据全量扫描把 IO 打满，所有看板查询超时——业务并没有算错，只是「查不到」。分开后，元数据服务故障只影响新建表、改权限、发新任务，已经部署的查询仍可跑。反例是权限服务：它虽然属于控制面，但在查询路径上做判定，必须加本地缓存与降级策略，不能简单当成「离线控制面」。
Q3 ODS → ADS 直连为什么是硬错误？什么情况下可以例外？
▸ 答案：它把本应集中在 DWD 的清洗、去重、状态过滤复制进每个报表，导致同一指标多套口径、上游字段变更时影响面不可枚举。例外只应发生在「一次性取数、不进生产 DAG、有到期日」的场景，并且必须在数据集清单里登记工单号与失效时间；没有到期日的例外等于永久放弃分层。
Q4 数据集清单里 owner / SLA / classification / retention 四项，各缺一项会导致什么具体后果？
▸ 答案：缺 owner → 口径争议和告警无人认领；缺 SLA → 无法区分「晚到」与「故障」，也没法做错误预算；缺分级 → 无法决定加密、脱敏、审批强度，合规审计说不清；缺留存 → 存储成本无上限，且「到期删除」没有依据。四项共同构成「这个数据集能不能被信任、能被谁用、能用多久」的最小契约。
Q5 延迟预算为什么必须用 P99 而不是平均值？
▸ 答案：数据链路的延迟分布是长尾的：平均值 20 分钟的链路，P99 可能是 3 小时（大促排队、上游补数、资源争抢）。业务承诺的是「最晚什么时候能看到」，对应 P99 或 P95；用平均值定预算，会在每个晚高峰违约，而且平均值会把少数极慢分区的影响完全抹掉。
Q6 一个 5 人的数据团队，日增 50 GB，是否应该自建湖仓？
▸ 答案：通常不该。自建湖仓的控制面（catalog、表格式维护、compaction、权限、元数据、调度）至少需要 1–2 人长期维护，5 人团队会被运维吃掉建模与业务分析产能。更合理的路径是托管数仓 / 托管湖仓（BigQuery、Snowflake、Databricks、MaxCompute）+ dbt + 托管 catalog，把控制面外包；等数据量或合规要求使托管成本超过自建总成本（含人力）时再迁移。
Q7 分层门禁上线，怎样避免「第一天全红、第二天全关」？
▸ 答案：分三阶段：① 只报告，把存量违规按 owner 分派并给豁免期；② 只阻断增量，本次变更新增的违规才失败，存量走白名单（带到期日）；③ 豁免到期后全量阻断。同时把规则数量控制在 4–6 条、每条都有可执行的修法，否则团队会把它当噪音，最终关掉门禁。