第25课：毕业项目-企业数据平台

🎉 毕业项目-企业数据平台

综合运用全部25课知识，构建完整企业级数据平台，覆盖采集→存储→处理→治理→服务全链路。

🐍 Python实战：核心实战

# 企业数据平台核心
import sqlite3, time, random, json, hashlib
from datetime import datetime, timedelta
from collections import defaultdict

# ===== 1. 数据采集层 =====
class IngestionLayer:
    def __init__(self): self.sources = {}; self.events = []
    def register(self, name, source_type): self.sources[name] = source_type
    def ingest(self, source_name, count):
        for i in range(count):
            self.events.append({
                'source': source_name,
                'event_id': hashlib.md5(f"{source_name}_{i}_{time.time()}".encode()).hexdigest()[:8],
                'data': {'value': round(random.uniform(10,1000),2)},
                'timestamp': datetime.now().isoformat()
            })
        return count

# ===== 2. 数据存储层 =====
class StorageLayer:
    def __init__(self):
        self.conn = sqlite3.connect(':memory:')
        self.conn.row_factory = sqlite3.Row
        self.tables = {}
    def create_table(self, name, schema):
        cols = ', '.join(f"{col} {dtype}" for col, dtype in schema.items())
        self.conn.execute(f"CREATE TABLE {name} ({cols})")
        self.tables[name] = schema
    def write(self, table, records):
        if not records: return 0
        cols = ', '.join(records[0].keys())
        vals_template = ', '.join(['?' for _ in records[0]])
        for r in records:
            self.conn.execute(f"INSERT INTO {table} ({cols}) VALUES ({vals_template})", list(r.values()))
        self.conn.commit()
        return len(records)
    def query(self, sql):
        return [dict(r) for r in self.conn.execute(sql).fetchall()]

# ===== 3. 数据处理层 =====
class ProcessingLayer:
    def __init__(self): self.pipelines = []
    def add_pipeline(self, name, fn): self.pipelines.append({'name':name,'fn':fn})
    def execute(self, data):
        for pipe in self.pipelines:
            data = pipe['fn'](data)
        return data

# ===== 4. 数据治理层 =====
class GovernanceLayer:
    def __init__(self): self.quality_rules = {}; self.lineage = defaultdict(list)
    def add_quality_rule(self, table, rule_fn): self.quality_rules[table] = rule_fn
    def add_lineage(self, source, target): self.lineage[source].append(target)
    def validate(self, table, records):
        if table in self.quality_rules:
            return self.quality_rules[table](records)
        return {'passed': True, 'total': len(records), 'failed': 0}

# ===== 5. 数据服务层 =====
class ServingLayer:
    def __init__(self): self.api_endpoints = {}
    def register(self, path, fn): self.api_endpoints[path] = fn
    def call(self, path, params=None):
        if path in self.api_endpoints:
            return self.api_endpoints[path](params or {})
        return {'error': 'not found'}

# ===== 组装企业数据平台 =====
class EnterpriseDataPlatform:
    def __init__(self, name):
        self.name = name
        self.ingestion = IngestionLayer()
        self.storage = StorageLayer()
        self.processing = ProcessingLayer()
        self.governance = GovernanceLayer()
        self.serving = ServingLayer()
    
    def bootstrap(self):
        # 注册数据源
        self.ingestion.register('mysql_orders', 'cdc')
        self.ingestion.register('api_products', 'api')
        self.ingestion.register('log_clicks', 'file')
        
        # 创建存储表
        self.storage.create_table('raw_orders', {'id':'INTEGER PRIMARY KEY','amount':'REAL','region':'TEXT','order_date':'TEXT'})
        self.storage.create_table('dim_region', {'id':'INTEGER PRIMARY KEY','name':'TEXT','zone':'TEXT'})
        self.storage.create_table('fct_daily_sales', {'id':'INTEGER PRIMARY KEY','order_date':'TEXT','region':'TEXT','total_amount':'REAL','order_count':'INTEGER'})
        
        # 维度数据
        regions = ['华东','华南','华北','西南','东北']
        for i, r in enumerate(regions):
            self.storage.write('dim_region', {'id':i+1,'name':r,'zone':f"Zone_{i+1}"})
        
        # 数据处理管道
        self.processing.add_pipeline('clean', lambda data: [r for r in data if r.get('amount',0) > 0])
        self.processing.add_pipeline('enrich', lambda data: [{**r,'order_date':datetime.now().strftime('%Y-%m-%d')} for r in data])
        
        # 质量规则
        self.governance.add_quality_rule('raw_orders', lambda records: {
            'passed': sum(1 for r in records if r.get('amount',0) > 0) == len(records),
            'total': len(records), 'failed': sum(1 for r in records if r.get('amount',0) <= 0)
        })
        
        # 血缘
        self.governance.add_lineage('mysql_orders','raw_orders')
        self.governance.add_lineage('raw_orders','fct_daily_sales')
        
        # API服务
        self.serving.register('/api/daily_sales', lambda p: self.storage.query(
            f"SELECT * FROM fct_daily_sales WHERE region = '{p.get('region','%')}' ORDER BY order_date"))
        self.serving.register('/api/quality', lambda p: self.governance.validate(p.get('table',''), []))
    
    def run_pipeline(self):
        # 1. 采集
        count = self.ingestion.ingest('mysql_orders', 1000)
        print(f"  采集: {count}条")
        
        # 2. 处理
        raw_records = [{'id':i+1,'amount':round(random.uniform(-10,5000),2),'region':random.choice(['华东','华南','华北','西南','东北'])} for i in range(1000)]
        processed = self.processing.execute(raw_records)
        print(f"  处理: {len(raw_records)}→{len(processed)}条")
        
        # 3. 存储
        written = self.storage.write('raw_orders', processed)
        print(f"  存储: {written}条")
        
        # 4. 质量检查
        quality = self.governance.validate('raw_orders', processed)
        print(f"  质量: {'通过' if quality['passed'] else '未通过'}")
        
        # 5. 聚合
        agg = defaultdict(lambda: {'total':0,'count':0})
        for r in processed:
            key = r['region']
            agg[key]['total'] += r.get('amount',0)
            agg[key]['count'] += 1
        for region, stats in agg.items():
            self.storage.write('fct_daily_sales', {
                'order_date': datetime.now().strftime('%Y-%m-%d'),
                'region': region,
                'total_amount': round(stats['total'],2),
                'order_count': stats['count']
            })
        
        return processed

# 运行平台
platform = EnterpriseDataPlatform("EnterpriseDataPlatform-v1")
platform.bootstrap()
print(f"企业数据平台: {platform.name}")
print("运行管道:")
result = platform.run_pipeline()

# 服务查询
print("\n数据服务:")
sales = platform.serving.call('/api/daily_sales', {'region':'华东'})
print(f"  /api/daily_sales?region=华东: {len(sales)}条")
for r in sales[:3]:
    print(f"    {r}")

# 血缘
print(f"\n数据血缘:")
for src, targets in platform.governance.lineage.items():
    for t in targets:
        print(f"  {src} → {t}")

print(f"\n指标: 采集源{len(platform.ingestion.sources)}个, 存储{len(platform.storage.tables)}表, 血缘{sum(len(v) for v in platform.governance.lineage.values())}条")
print("✅ 验证通过 - 企业数据平台运行正常")

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

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

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

📚 延伸阅读与参考