60.4 元数据管理与数据目录
一句话
元数据是数据平台的「控制面数据库」：它不存业务数据，却决定谁能发现、能否信任、改动会炸到谁；没有元数据的平台，只是一个更贵的网盘。
三类元数据，三种生产者
| 类型 | 内容 | 谁生产 | 谁消费 | 更新方式 |
|---|---|---|---|---|
| 技术元数据 | 库 / 表 / 列 / 类型 / 分区 / 文件 / 行数 / min-max / 存储位置 | 采集器从 Metastore、JDBC、文件系统抓 | 查询引擎、调度、质量 | 定时爬取 + 事件 |
| 业务元数据 | 指标定义、术语、owner、分级、标签、血缘、SLA | 人（数据开发 / 分析 / 安全） | 分析师、BI、审计 | 声明式（代码 / 表单） |
| 操作元数据 | 任务、运行、耗时、失败、质量结果、访问日志、成本 | 调度器、引擎、质量平台、审计 | on-call、FinOps、安全 | 事件流 |
🔑 三类元数据的新鲜度要求完全不同：技术元数据可以分钟级，业务元数据按变更（可能几周），操作元数据必须秒级。用一套采集周期覆盖三者，要么太慢，要么把系统压垮。
采集：五种方式，五种坑
| 方式 | 延迟 | 覆盖率 | 成本 | 典型坑 |
|---|---|---|---|---|
| JDBC / API 爬取 | 分钟–小时 | 高（有库就能抓） | 中（全量扫描） | 大库 information_schema 慢；拿不到视图/存储过程逻辑 |
| Metastore 事件 | 秒级 | 中（只覆盖 Hive/Glue 等） | 低 | 事件丢一条，元数据永久缺失；要定期对账 |
| 查询日志解析 | 分钟 | 高（能拿到真实使用） | 高（日志量大） | 只能看到「被查询的」，看不到「建了没用的」 |
| SDK / OpenLineage | 实时 | 取决于接入率 | 低（应用埋点） | 各引擎接入不一致；字段规范不统一 |
| 手工声明 | 天–周 | 低但最准 | 人力 | 三个月后过期，与真实链路脱节 |
生产上的组合是：技术元数据自动爬 + 血缘事件推送 + 业务元数据代码化 + 每周对账。表级血缘的图算法与爆炸半径示例见 46.5 数据治理与血缘。
# 业务元数据代码化：和模型放在同一个仓库，随 PR 评审
dataset: dwd_trade_order_di
owner: dw-trade
layer: DWD
classification: L3
sla:
freshness_minutes: 15
available_by: "06:00"
retention_days: 1095
columns:
- name: amount_cents
description: 订单金额（分），含运费，不含退款
classification: L3
⚠️ 采集失败不会让业务挂，但会让「影响面」失真——改表时漏掉一个下游，比没有血缘更危险。所以采集器本身要有 freshness 指标：上次成功时间、本次新增/删除的表数、解析失败数。
数据目录：搜索只是入口
| 能力 | 解决什么 | 没有它会怎样 |
|---|---|---|
| 搜索 / 发现 | 「有没有这张表」 | 重复建表，同一数据 N 份 |
| 详情 / schema | 「字段什么意思」 | 猜字段、用错列 |
| owner / 值班 | 「出问题找谁」 | 告警发到群里没人接 |
| 分级 / 标签 | 「能不能用、要不要脱敏」 | 敏感数据被当普通数据开放 |
| 血缘 / 影响面 | 「改它炸到谁」 | 上线后下游报表全空 |
| 质量分 / 最近产出 | 「这数能不能信」 | 拿三天前的数据做决策 |
| 用量 / 成本 | 「谁在用、值不值」 | 冷表长期占存储，没人敢删 |
| 权限申请 | 「怎么合法拿到访问权」 | 私下导数据，审计不可见 |
🔑 目录的核心价值不是「搜索框」，而是变更前的检查清单与口径争议时的证据链。搜索谁都会做；难的是「点开一张表，能看到它的上游、下游、owner、质量、成本和审批入口」。
指标字典：把口径写进代码
| 字段 | 必须写清 | 反例 |
|---|---|---|
| 名称 | 业务唯一名，不用缩写 | gmv2、amt |
| 定义 | 一句话说清业务含义 | 「成交额」 |
| 公式 | 可执行的表达式 | 「大概就是销售额」 |
| 维度 | 可按什么拆 | 只写「按天」 |
| 过滤 | 含哪些状态、哪些渠道 | 不写「是否含退款」 |
| 来源 | 哪张表、哪一列 | 「来自订单表」 |
| owner | 谁签字、谁维护 | 空 |
| 版本 | 口径变更记录 | 直接改，不留痕 |
指标平台把这份字典变成可执行的 SQL / 语义层（如 dbt metrics、Cube、MetricFlow）；指标最终落在哪一层、由谁消费，见 60.2 数仓分层 (ODS / DWD / DWS / ADS)。好处不是「自动生成 SQL」，而是口径变更走 PR：谁改的、影响哪些报表、什么时候生效，全部可查。
血缘：表级回答影响面，列级回答口径
| 问题 | 表级血缘 | 列级血缘 |
|---|---|---|
| 改这张表会影响哪些任务 | ✅ | ✅（更细） |
| 这个指标来自哪张表的哪一列 | ❌ | ✅ |
| 两个报表同名指标为什么不同 | ❌ | ✅（不同来源列/过滤） |
| 删除这一列安全吗 | ❌ | ✅ |
列级血缘的实现难度在 SQL 解析：
SELECT *、CTE、子查询、窗口函数、UDF、动态 SQL、临时表，都会让「列从哪来」变得不确定；- 生产级解析器用 AST（sqlglot、Apache Calcite），不是正则；
- 解析失败时必须 fail-closed：不发布这条血缘、告警、要求补声明，而不是猜一个边。
代码：爬 schema、建血缘、算影响面、发事件
#!/usr/bin/env python3
"""元数据爬取 + 血缘 + 影响面（SQLite，无第三方依赖）。"""
import json
import re
import sqlite3
import tempfile
import uuid
from pathlib import Path
DB = Path(tempfile.mkdtemp(prefix="meta-")) / "dw.db"
DDL = """
CREATE TABLE ods_orders (id TEXT, user_id TEXT, amount INTEGER, status TEXT, dt TEXT);
CREATE TABLE ods_users (id TEXT, phone TEXT, city TEXT);
CREATE TABLE dwd_orders (order_id TEXT, user_id TEXT, amount INTEGER, status TEXT, dt TEXT);
CREATE TABLE dwd_users (user_id TEXT, phone TEXT, city TEXT);
CREATE TABLE dws_sales (dt TEXT, gmv INTEGER, orders INTEGER);
CREATE TABLE ads_gmv (dt TEXT, gmv INTEGER);
CREATE TABLE tmp_export (x TEXT);
INSERT INTO ods_orders VALUES ('o1','u1',100,'paid','2026-10-01'),
('o2','u2',200,'paid','2026-10-01');
INSERT INTO ods_users VALUES ('u1','13800000001','北京'),('u2','13800000002','上海');
"""
TASKS = [
{"name": "dwd_orders_daily",
"sql": "INSERT INTO dwd_orders SELECT id,user_id,amount,status,dt FROM ods_orders",
"columns": {"order_id": "ods_orders.id", "user_id": "ods_orders.user_id",
"amount": "ods_orders.amount", "status": "ods_orders.status", "dt": "ods_orders.dt"}},
{"name": "dwd_users_daily",
"sql": "INSERT INTO dwd_users SELECT id,phone,city FROM ods_users",
"columns": {"user_id": "ods_users.id", "phone": "ods_users.phone", "city": "ods_users.city"}},
{"name": "dws_sales_daily",
"sql": """INSERT INTO dws_sales
SELECT o.dt, SUM(o.amount), COUNT(*)
FROM dwd_orders o JOIN dwd_users u ON o.user_id = u.user_id GROUP BY o.dt""",
"columns": {"dt": "dwd_orders.dt", "gmv": "dwd_orders.amount", "orders": "dwd_orders.order_id"}},
{"name": "ads_gmv_daily",
"sql": "INSERT INTO ads_gmv SELECT dt,gmv FROM dws_sales",
"columns": {"dt": "dws_sales.dt", "gmv": "dws_sales.gmv"}},
# 解析器不支持的写法：CTE。生产上应交给 sqlglot/Calcite，解析失败要拒绝发布血缘
{"name": "dwd_orders_merge",
"sql": "WITH x AS (SELECT * FROM ods_orders) INSERT INTO dwd_orders SELECT * FROM x",
"columns": {}},
]
META = {
"ods_orders": {"owner": "data-eng", "level": "L3", "desc": "订单源表"},
"ods_users": {"owner": "data-eng", "level": "L3", "desc": "用户源表"},
"dwd_orders": {"owner": "dw-trade", "level": "L3", "desc": "订单明细"},
"dwd_users": {"owner": "dw-user", "level": "L3", "desc": "用户明细"},
"dws_sales": {"owner": "dw-trade", "level": "L2", "desc": "销售日汇总"},
"ads_gmv": {"owner": "bi-team", "level": "L2", "desc": "GMV 看板"},
"tmp_export": {"owner": "", "level": "", "desc": ""},
}
COL_META = {"phone": "L4", "id": "L2", "amount": "L3", "gmv": "L2"}
def crawl(con):
out = {}
for (name,) in con.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"):
cols = [r[1] for r in con.execute(f"PRAGMA table_info({name})")]
rows = con.execute(f"SELECT COUNT(*) FROM {name}").fetchone()[0]
out[name] = {"columns": cols, "rows": rows}
return out
def parse(sql):
if re.search(r"\bWITH\b", sql, re.I) or re.search(r"\bFROM\s*\(", sql, re.I):
raise ValueError("含 WITH/子查询，当前解析器不支持")
clean = re.sub(r"--[^\n]*", "", sql)
target = re.search(r"INSERT\s+INTO\s+([A-Za-z_]\w*)", clean, re.I)
sources = re.findall(r"\b(?:FROM|JOIN)\s+([A-Za-z_]\w*)", clean, re.I)
if not target:
raise ValueError("找不到 INSERT INTO 目标表")
return target.group(1), list(dict.fromkeys(sources))
def main():
con = sqlite3.connect(DB)
con.executescript(DDL)
tables = crawl(con)
print("=== 爬取到的数据集（技术元数据）===")
print(f"{'name':<12}{'rows':>5} columns")
for n, t in tables.items():
print(f"{n:<12}{t['rows']:>5} {','.join(t['columns'])}")
edges, col_edges, errors = set(), set(), []
for task in TASKS:
try:
out_table, sources = parse(task["sql"])
for s in sources:
edges.add((s, out_table))
for tgt_col, src in task["columns"].items():
col_edges.add((src, f"{out_table}.{tgt_col}"))
except ValueError as e:
errors.append((task["name"], str(e)))
print("\n=== 表级血缘 ===")
for s, t in sorted(edges):
print(f" {s} -> {t}")
def downstream(start, graph):
seen, stack = [], [start]
while stack:
cur = stack.pop()
for s, t in graph:
if s == cur and t not in seen:
seen.append(t)
stack.append(t)
return seen
impact = downstream("ods_orders", edges)
col_impact = downstream("ods_orders.amount", col_edges)
print(f"\n=== 改 ods_orders 的影响面 ===\n {len(impact)} 个下游：{impact}")
print(f"\n=== 改 ods_orders.amount 的列级影响 ===\n {len(col_impact)} 个下游：{col_impact}")
total_cols = sum(len(t["columns"]) for t in tables.values())
owned = sum(1 for n in tables if META.get(n, {}).get("owner"))
desc = sum(1 for n in tables if META.get(n, {}).get("desc"))
level = sum(1 for n in tables if META.get(n, {}).get("level"))
classified = sum(1 for t in tables.values() for c in t["columns"] if c in COL_META)
covered = {s for s, _ in edges} | {t for _, t in edges}
print(f"\n=== 覆盖率 ===")
print(f" owner {owned}/{len(tables)}；描述 {desc}/{len(tables)}；分级 {level}/{len(tables)}")
print(f" 列分级 {classified}/{total_cols}；血缘覆盖 {len(covered)}/{len(tables)}"
f"（孤立表：{set(tables)-covered}）")
print(f"\n=== 解析失败（fail-closed，不发布该任务血缘）===")
for name, why in errors:
print(f" {name}: {why}")
event = {
"eventType": "COMPLETE",
"eventTime": "2026-10-30T21:30:00+08:00",
"producer": "https://example.com/metadata-crawler",
"run": {"runId": str(uuid.uuid5(uuid.NAMESPACE_URL, "dws_sales_daily/2026-10-30"))},
"job": {"namespace": "dw", "name": "dws_sales_daily"},
"inputs": [{"namespace": "sqlite://dw", "name": "dwd_orders"},
{"namespace": "sqlite://dw", "name": "dwd_users"}],
"outputs": [{"namespace": "sqlite://dw", "name": "dws_sales"}],
}
print("\n=== OpenLineage 风格事件（字段有裁剪）===")
print(json.dumps(event, ensure_ascii=False, indent=1))
main()
本机实测（macOS，Apple M1，Python 3.14.8）：
=== 爬取到的数据集（技术元数据）===
name rows columns
ads_gmv 0 dt,gmv
dwd_orders 0 order_id,user_id,amount,status,dt
dwd_users 0 user_id,phone,city
dws_sales 0 dt,gmv,orders
ods_orders 2 id,user_id,amount,status,dt
ods_users 2 id,phone,city
tmp_export 0 x
=== 表级血缘 ===
dwd_orders -> dws_sales
dwd_users -> dws_sales
dws_sales -> ads_gmv
ods_orders -> dwd_orders
ods_users -> dwd_users
=== 改 ods_orders 的影响面 ===
3 个下游：['dwd_orders', 'dws_sales', 'ads_gmv']
=== 改 ods_orders.amount 的列级影响 ===
3 个下游：['dwd_orders.amount', 'dws_sales.gmv', 'ads_gmv.gmv']
=== 覆盖率 ===
owner 6/7；描述 6/7；分级 6/7
列分级 8/22；血缘覆盖 6/7（孤立表：{'tmp_export'}）
=== 解析失败（fail-closed，不发布该任务血缘）===
dwd_orders_merge: 含 WITH/子查询，当前解析器不支持
=== OpenLineage 风格事件（字段有裁剪）===
{
"eventType": "COMPLETE",
"eventTime": "2026-10-30T21:30:00+08:00",
"producer": "https://example.com/metadata-crawler",
"run": {
"runId": "19ae45b7-e2a5-50f7-afad-f0a8057f7e34"
},
"job": {
"namespace": "dw",
"name": "dws_sales_daily"
},
"inputs": [
{
"namespace": "sqlite://dw",
"name": "dwd_orders"
},
{
"namespace": "sqlite://dw",
"name": "dwd_users"
}
],
"outputs": [
{
"namespace": "sqlite://dw",
"name": "dws_sales"
}
]
}
读数：
| 结果 | 含义 | 生产动作 |
|---|---|---|
| 表级影响 3 个下游 | 改 ods_orders 要通知 DWD / DWS / ADS | 上线前自动 @ owner，排期 |
| 列级影响 3 跳 | amount → dwd_orders.amount → dws_sales.gmv → ads_gmv.gmv | 改列前能指出「动的是 GMV 口径」 |
| 血缘覆盖 6/7 | tmp_export 是孤立表，无上游也无下游 | 下线或补登记，别让它变成「谁也不敢删」 |
| 列分级 8/22 | 大量列没有分级 | 无法做列级权限与脱敏 |
| 解析失败 1 条 | CTE 不在解析器能力内，血缘缺失 | 换 AST 解析器，或要求任务补声明 |
🔑 列级血缘的价值在这条链上：
ods_orders.amount一改，能直接指出「GMV 看板的 gmv 会变」。表级血缘只能说「ads_gmv 可能受影响」，给不出口径证据。
元数据本身也要治理
| 问题 | 指标 | 阈值 / 动作 |
|---|---|---|
| 缺 owner | owner 覆盖率 | < 95% 不允许发布到生产目录 |
| 缺描述 | 描述覆盖率 | 核心表 100%，长尾 ≥ 80% |
| 缺分级 | 分级覆盖率 | L3/L4 必须 100% |
| 血缘缺失 | 血缘覆盖率 | 核心链路 100%，解析失败必须告警 |
| 孤立表 | 无上游且无下游 | 季度清理，先软删再物理删 |
| 元数据过期 | 上次采集时间 | 超过 24h 告警 |
| 元数据泄露 | 列名/注释含敏感信息 | 目录本身要做权限与审计 |
分级字段的判定口径见 38.5 数据分级与脱敏。元数据的权限常被忽略：列名、注释、样本数据、owner 邮箱本身就是情报。一个能匿名搜索「身份证」列名的人，已经获得了攻击地图。目录要按分级授权，查询行为要审计（见 3.5 审计与取证）；目录本身的访问策略与数据权限共用一套模型，见 60.6 数据安全与权限。
生产落地：指名 → 可抄 → 可验 → 可退
指名：目录用 DataHub（替代 OpenMetadata / Apache Atlas）；血缘事件用 OpenLineage + Marquez（替代自研埋点）；转换层血缘用 dbt docs / dbt lineage；云上可直接用 Unity Catalog / AWS Glue Catalog。
可抄（DataHub 采集 PostgreSQL，未验证：本机未安装 DataHub，GMS 默认 8080、前端 9002）：
# recipe.yml
source:
type: postgres
config:
host: localhost
port: 5432
username: meta_reader
password: ${PG_PASSWORD}
database: shop
include_tables: true
include_views: true
sink:
type: datahub-rest
config:
server: "http://localhost:8080"
datahub ingest -c recipe.yml
# 预期：Produced 42 entities / 0 failures（数字随库而定）
# OpenLineage 事件推给 Marquez（默认 API 5000）
curl -X POST http://localhost:5000/api/v1/lineage \
-H 'Content-Type: application/json' \
-d @event.json
# 预期：201 Created
可验：本节脚本直接可跑，输出影响面与覆盖率；产品侧在目录里搜索 dwd_trade_order_di，应能看到 owner、schema、上下游与最近一次采集时间。可退：采集器是只读旁路，停掉不影响业务；血缘发布失败只影响「新链路不可见」，不应阻断任务，但要告警并在发布门禁里检查「核心表血缘是否齐全」。
反模式
| 反模式 | 后果 |
|---|---|
| 血缘靠人工维护文档 | 三个月后与真实链路脱节，改表仍靠问人 |
| 只采集技术元数据，不登记业务口径 | 目录能搜到表，但没人知道字段能不能用 |
| 解析失败就「跳过」 | 血缘缺一条边，影响面少一半，事故时才发现 |
| 用正则解析所有 SQL | CTE / 子查询 / UDF 全漏，列级血缘不可信 |
| 采集器没有 freshness 监控 | 元数据停在三个月前，目录变成「历史博物馆」 |
| 目录不做权限 | 列名与注释泄露敏感信息，等于给出攻击地图 |
| 指标改名不留别名 | 老报表引用断裂，搜索不到历史指标 |
| 只建目录不接权限申请 | 用户找到数据却拿不到权限，最后还是私下导 |
自检
Q1 技术元数据、业务元数据、操作元数据，各自由谁生产？为什么不能都用「定时爬取」？
▸ 答案：技术元数据由采集器从 Metastore / JDBC / 文件系统生产；业务元数据由数据开发、分析、安全人员声明（代码或表单）；操作元数据由调度器、查询引擎、质量平台、审计系统以事件方式产生。定时爬取只能拿到「结构」，拿不到「口径」与「运行状态」；操作元数据要求秒级，爬取周期太慢。
Q2 为什么「Metastore 事件 + 定期对账」要一起用，只留一个会怎样？
▸ 答案：事件延迟低，但会丢（网络、重启、消费失败），丢了就永久缺失；定期对账能补齐，但延迟高。只留事件 → 元数据缺口无人发现；只留对账 → 新建的表几小时内搜不到，影响开发效率。两者组合：事件做实时， 对账做兜底，并监控「事件消费延迟」与「对账新增/删除数」。
Q3 表级血缘已经能算影响面，为什么还要列级血缘？举一个只有列级能回答的问题。
▸ 答案：表级只能说「这张表的下游是谁」，回答不了「这个指标用了哪一列、经过什么过滤」。例如两个报表都叫「销售额」，一个用 amount，另一个用 amount - refund；表级血缘显示它们都来自 dwd_orders，看起来一致，实际口径不同。只有列级血缘能追到「一个来自 amount，一个来自 amount 与 refund 的组合」，给出差异证据。
Q4 SQL 解析器遇到 CTE 解析失败，为什么必须 fail-closed，而不是「先不给血缘、静默跳过」？
▸ 答案：静默跳过会让血缘图缺边，而缺边的影响面看起来「更小」——改表时系统告诉你「没有下游」，实际下游存在。这比没有血缘更危险，因为人会更相信系统。正确做法是标记该任务血缘为 UNRESOLVED、告警、在发布门禁里阻断核心链路，直到补声明或换解析器。
Q5 数据目录的核心价值为什么不是「搜索」？
▸ 答案：搜索只解决「找到」，但数据平台真正的问题是「找到后敢不敢用、改它会不会炸、出事找谁」。这些要靠 owner、分级、血缘、质量分、成本、审批入口共同回答。一个只有搜索的目录，最终会变成「能搜到但没人信」的页面，用户还是回去问同事。
Q6 元数据本身为什么要做权限与审计？
▸ 答案：元数据包含列名、注释、样本、owner 联系方式、血缘与用量，这些本身就是攻击地图与商业情报。能搜索「身份证 / 银行卡」列的人，已经知道去哪偷；能看血缘的人，知道哪个上游最薄弱。目录要按数据分级授权，查询与导出要审计，敏感列的注释也要脱敏。
Q7 一个可用的指标字典，至少写清哪几项？
▸ 答案：名称、业务定义、可执行公式、维度、过滤条件（含不含退款/取消）、来源表与列、owner、版本与变更记录。缺公式就无法复现；缺过滤就会同名不同值；缺 owner 就没人对口径负责；缺版本就无法回答「这个数什么时候变的」。