元数据卡
- 前置知识：第1-2章（数据生命周期、数据清洗）
- 预计时间：40 分钟
- 核心难度：进阶
- 阅读模式：轻松漫游
- 完成标志：能设计数据质量维度框架、搭建基础的数据目录和监控
你的进度
数据伦理让你意识到：数据本身是资产，也是风险。预言厅里堆积的数据越来越多，但没人能回答一个基本问题：哪些数据是可信的？谁在用它们？它们应该被保存多久？
你翻开档案室的规章手册，发现里面写满了规则——数据质量、数据目录、数据生命周期策略。这不是最性感的主题，但缺少了它，你手里的数据就是一堆有风险的数字。
你的任务
你的团队每天处理 100 个数据管道。今天早上有个管道突然空了，导致下游报告全部报错。你觉得——应该有人盯着这些管道。但人盯不过来。数据治理就是一套体系化的做法：定义数据质量、建立目录、自动监控、确保数据可信可用。
数据治理不是什么
数据治理不是"加一堆审批流程让数据分析师没法干活"。它是基础设施建设——和代码 CI/CD 一样自然的工程实践。你希望数据管道出了问题能自动发现，而不是两周后业务方说"报告数据不对"。
数据质量六维度
业界通用的数据质量框架定义六个维度：
| 维度 | 定义 | 检查方法 |
|---|---|---|
| 完整性 | 数据是否缺失 | df.isnull().sum() |
| 准确性 | 数据是否反映真实情况 | 校验规则（如: 年龄 > 0） |
| 一致性 | 同一数据在不同系统中是否一致 | 跨系统对账 |
| 时效性 | 数据是否在预期时间内可用 | 检查延迟时间 |
| 唯一性 | 是否存在重复记录 | df.duplicated().sum() |
| 有效性 | 数据是否符合格式规范 | 正则校验、枚举值校验 |
# 数据质量检查函数 —— 基础版
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
def quality_check(df, table_name):
checks = {}
# 完整性
missing = df.isnull().sum()
checks["missing_rate"] = (missing / len(df)).to_dict()
# 准确性 —— 示例规则
if "age" in df.columns:
checks["invalid_age"] = int((df["age"] < 0).sum())
if "email" in df.columns:
checks["invalid_email"] = int(
(~df["email"].str.contains("@", na=False)).sum()
)
# 唯一性
checks["duplicate_rate"] = df.duplicated().mean()
# 时效性
if "load_timestamp" in df.columns:
max_age = (datetime.now() - df["load_timestamp"].max()).total_seconds()
checks["data_max_age_seconds"] = max_age
return pd.Series(checks, name=table_name)
# 对每个表运行检查
results = pd.DataFrame([
quality_check(df_missions, "missions"),
quality_check(df_logs, "logs"),
])
print(results)每个团队的数据质量规则不同。你的任务是定义关键规则并用代码执行检查，而不是靠人肉排查。
数据目录
数据多了以后，没人记得"mission_status 表在哪里""total_resources 字段的含义是什么"。数据目录（Data Catalog）帮你回答这些问题。
一个基础的数据目录记录每张表的：
# 构建数据目录的结构
catalog_entries = []
for table_name, df in {"missions": df_missions, "logs": df_logs}.items():
for col in df.columns:
entry = {
"table": table_name,
"column": col,
"dtype": str(df[col].dtype),
"non_null_count": df[col].notna().sum(),
"unique_values": df[col].nunique(),
"sample_values": df[col].dropna().unique()[:3].tolist(),
"definition": "", # 人工填写
"owner": "", # 人工填写
}
catalog_entries.append(entry)
catalog = pd.DataFrame(catalog_entries)
catalog.to_csv("data_catalog.csv", index=False)目录中最关键的信息是definition（字段含义）和owner（谁负责维护）。技术信息可以从数据中自动提取，语义信息需要人维护。
数据监控
质量检查和目录提供了静态视图。但数据变化是持续的——你需要监控新加载数据是否突然变差。
# 监控基线 —— 以行为数为例
from collections import deque
history = deque(maxlen=30) # 记录最近30天的行数
def monitor_row_count(table_name, current_count, history):
"""如果当前行数偏离基线超过 3 个标准差，触发告警"""
if len(history) < 7:
history.append(current_count)
return # 基线建立期，不告警
mean_ = np.mean(history)
std_ = np.std(history)
if abs(current_count - mean_) > 3 * std_:
print(f"ALERT: {table_name} row count anomaly!")
print(f" baseline: {mean_:.0f} +/- {3*std_:.0f}")
print(f" current: {current_count}")
return True
history.append(current_count)
return False上述是最简版本的监控：行数波动。实际场景中你还要监控缺失率、分布变化、新鲜度（数据多久没更新了）。
SLAs 和服务等级
数据治理的核心产出：每张表有一个 SLA 记录。
- 数据可用时间：每天上午 8:00 前数据应可用
- 最低质量要求：缺失率 < 5%，重复率 < 1%
- 延迟上限：数据延迟不超过 6 小时
- 告警响应时效：P0 级别问题 1 小时内响应
# 记录 SLA
sla_dashboard = pd.DataFrame({
"table": ["missions", "logs", "reports"],
"availability_time": ["08:00", "07:00", "09:00"],
"max_missing_rate": [0.05, 0.03, 0.01],
"max_duplicate_rate": [0.01, 0.01, 0.00],
"max_lag_hours": [6, 4, 12],
"p0_response_minutes": [60, 60, 120],
})常见陷阱
- 一次性清理，没有持续监控。数据质量下降是持续的——你今天修好了，明天可能又坏了。
- 质量规则太苛刻。拒绝率 100% 的数据比 90% 质量的数据更糟。先接受，标记异常，再逐步提高标准。
- 目录建好就没人管了。目录应该持续更新——有自动化导入，也有人工维护的渠道。
- 告警太多变成噪音。每次小波动都发邮件，一周后没人看告警。要分层：P0 电话，P1 邮件，P2 仪表盘。
通关挑战
- 热身：选一个你常用的数据集，列出它的六个数据质量维度评分。
- 挑战：为一组数据管道搭建一个完整的质量监控系统：定义检查规则 → 建立基线 → 每日运行检查 → 告警通知。
- 排障：一个管道反复触发"行数过少"告警。你怎么找出是数据真的少了还是代码写错了？
验收标准
- 能定义并应用数据质量六维度
- 能建立基础的数据目录
- 能搭建简单的数据质量监控
- 知道如何分层设置告警阈值
旅人笔记
数据质量不是一次性工程。你需要持续检查、持续改进——就像测试代码一样。
下一站预告
知道数据有没有问题之后，下一个问题是：数据从哪里来、经过什么变换、影响了什么下游？下一章，数据血缘与元数据。