第24课：数据治理平台

🏛️ 数据治理平台

构建企业级数据治理平台，集成数据目录、质量监控、血缘追踪、权限管理。

🐍 Python实战：核心实战

# 数据治理平台
import hashlib, json, time, random
from datetime import datetime
from collections import defaultdict
from dataclasses import dataclass, field

@dataclass
class DataAsset:
    name: str; owner: str; tier: str; pii_fields: list = field(default_factory=list)
    tags: list = field(default_factory=list); quality_score: float = 0
    access_count: int = 0; created: str = ""

class GovernancePlatform:
    def __init__(self):
        self.catalog = {}; self.policies = []; self.audit_log = []
        self.access_matrix = defaultdict(lambda: defaultdict(set))
    def register(self, asset):
        self.catalog[asset.name] = asset
        return asset
    def add_policy(self, name, rule_type, params):
        self.policies.append({'name':name,'type':rule_type,'params':params})
    def check_access(self, user, asset_name, action):
        asset = self.catalog.get(asset_name)
        if not asset: return False, "Asset not found"
        # RBAC check
        user_roles = self.access_matrix[user]
        for policy in self.policies:
            if policy['type'] == 'rbac':
                if action in policy['params'].get('denied_actions',[]):
                    if user in policy['params'].get('denied_users',[]):
                        self.audit(f"DENY {user} {action} {asset_name}")
                        return False, f"Policy {policy['name']} denied"
        # PII check
        if action == 'read' and asset.pii_fields:
            self.audit(f"PII_ACCESS {user} {asset_name} fields={asset.pii_fields}")
        self.audit(f"ALLOW {user} {action} {asset_name}")
        return True, "Access granted"
    def mask_pii(self, records, fields):
        masked = []
        for r in records:
            r = r.copy()
            for f in fields:
                if f in r and isinstance(r[f], str) and len(r[f]) > 2:
                    r[f] = r[f][:1] + '*' * (len(r[f])-2) + r[f][-1:]
            masked.append(r)
        return masked
    def audit(self, message):
        self.audit_log.append({'message':message,'time':datetime.now().isoformat()})
    def quality_dashboard(self):
        scored = [a for a in self.catalog.values() if a.quality_score > 0]
        return {
            'total_assets': len(self.catalog),
            'avg_quality': round(sum(a.quality_score for a in scored)/len(scored),2) if scored else 0,
            'pii_assets': sum(1 for a in self.catalog.values() if a.pii_fields),
            'total_access': sum(a.access_count for a in self.catalog.values()),
        }

# 构建治理平台
gov = GovernancePlatform()
gov.register(DataAsset('customers','data-team','gold',['phone','email'],['customer','pii'],0.95,100))
gov.register(DataAsset('orders','data-team','silver',[],['order','transaction'],0.88,500))
gov.register(DataAsset('products','product-team','gold',[],['product','catalog'],0.92,200))

gov.add_policy('pii_protection','rbac',{'denied_actions':['export'],'denied_users':['analyst']})

# 测试
allowed, msg = gov.check_access('analyst','customers','read')
print(f"analyst读customers: {allowed} ({msg})")
allowed, msg = gov.check_access('analyst','customers','export')
print(f"analyst导出customers: {allowed} ({msg})")

# 数据脱敏
records = [{'name':'张三','phone':'13800138000','email':'zhang@test.com','amount':1000}]
masked = gov.mask_pii(records, ['phone','email'])
print(f"脱敏: {masked}")

print(f"\n治理仪表盘: {gov.quality_dashboard()}")
print(f"审计日志: {len(gov.audit_log)}条")
print("✅ 验证通过 - 数据治理平台运行正常")

📖 原理深入：企业数据平台架构

🏆 最佳实践：数据平台建设检查清单

🔬 深度案例：生产环境实战经验

📚 延伸阅读与参考

🤔 常见问题与解答