数据工程实战课程 · 数据存储阶段
元数据是描述数据的数据，有效元数据管理是数据治理的基石。
# 元数据目录服务
import json, time
from datetime import datetime
from collections import defaultdict
from dataclasses import dataclass, field
@dataclass
class TableMeta:
name: str; database: str; owner: str; description: str = ""
tier: str = "silver"; tags: list = field(default_factory=list)
upstream: list = field(default_factory=list); downstream: list = field(default_factory=list)
quality_score: float = 0.0; row_count: int = 0; columns: list = field(default_factory=list)
class MetadataCatalog:
def __init__(self):
self.tables = {}; self.tag_index = defaultdict(set); self.owner_index = defaultdict(set)
self.lineage = defaultdict(lambda: {'up':set(),'down':set()})
def register(self, meta):
key = f"{meta.database}.{meta.name}"
self.tables[key] = meta
for tag in meta.tags: self.tag_index[tag].add(key)
self.owner_index[meta.owner].add(key)
for up in meta.upstream:
self.lineage[key]['up'].add(up)
self.lineage[up]['down'].add(key)
def search(self, tag=None, owner=None):
candidates = set(self.tables.keys())
if tag: candidates &= self.tag_index.get(tag, set())
if owner: candidates &= self.owner_index.get(owner, set())
return [self.tables[k] for k in candidates]
def get_lineage(self, key, direction='both', depth=3):
visited = set(); result = {'upstream':[],'downstream':[]}
def traverse(k, d, dep):
if dep > depth or k in visited: return
visited.add(k)
if d in ('upstream','both'):
for up in self.lineage[k]['up']: result['upstream'].append({'table':up,'depth':dep}); traverse(up,'upstream',dep+1)
if d in ('downstream','both'):
for dn in self.lineage[k]['down']: result['downstream'].append({'table':dn,'depth':dep}); traverse(dn,'downstream',dep+1)
traverse(key, direction, 1)
return result
def impact(self, key):
lin = self.get_lineage(key, 'downstream')
return {'source':key,'affected':len(set(i['table'] for i in lin['downstream']))}
def quality_report(self):
scored = [t for t in self.tables.values() if t.quality_score > 0]
return {'total':len(self.tables),'avg_quality':round(sum(t.quality_score for t in scored)/len(scored),2) if scored else 0,'no_owner':sum(1 for t in self.tables.values() if not t.owner)}
cat = MetadataCatalog()
cat.register(TableMeta('raw_orders','ods','data-team','原始订单','raw',['orders','raw'],[],[],0.85,100000))
cat.register(TableMeta('clean_orders','dwd','data-team','清洗订单','silver',['orders','cleaned'],['ods.raw_orders'],[],0.92,95000))
cat.register(TableMeta('daily_sales','dws','bi-team','日销售汇总','gold',['sales','agg'],['dwd.clean_orders'],[],0.95,365))
cat.register(TableMeta('sales_dashboard','ads','bi-team','销售大屏','gold',['dashboard'],['dws.daily_sales'],[],0.98,12))
print("元数据目录:")
print(f" 统计: {len(cat.tables)}表, 质量{cat.quality_report()}")
print(f" 搜索(tag=sales): {[t.name for t in cat.search(tag='sales')]}")
lin = cat.get_lineage('ods.raw_orders')
print(f" 血缘(raw_orders): 下游{[i['table'] for i in lin['downstream']]}")
print(f" 影响分析(raw_orders): {cat.impact('ods.raw_orders')}")
print("✅ 验证通过 - 元数据目录运行正常")🎁 下一课预告：流处理概述！
| 策略 | 原理 | 节省比例 |
|---|---|---|
| 分层存储 | 热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、数据产品规划 |
| 数据管家 | 数据质量与治理 | 质量规则、数据字典 |