第07课：数据湖架构

🌊 数据湖架构

数据湖以原始格式存储大量结构化和非结构化数据，采用Schema-on-Read模式。

🐍 Python实战：核心实战

# 数据湖存储引擎
import json, hashlib, time, random
from datetime import datetime
from collections import defaultdict
from dataclasses import dataclass, field

@dataclass
class LakeTable:
    name: str; zone: str; format: str; schema: dict = field(default_factory=dict)
    partitions: list = field(default_factory=list); record_count: int = 0; size_bytes: int = 0
    created_at: str = ""; updated_at: str = ""

class DataLakeEngine:
    def __init__(self, name):
        self.name = name
        self.zones = {'raw':{'tables':{}},'silver':{'tables':{}},'gold':{'tables':{}}}
        self.catalog = {}
    def register_table(self, zone, name, fmt, schema=None, partitions=None):
        t = LakeTable(name, zone, fmt, schema or {}, partitions or [], created_at=datetime.now().isoformat(), updated_at=datetime.now().isoformat())
        self.zones[zone]['tables'][name] = t; self.catalog[name] = t; return t
    def write_data(self, name, records):
        if name not in self.catalog: raise ValueError(f"表{name}未注册")
        t = self.catalog[name]
        t.record_count += len(records)
        t.size_bytes += len(json.dumps(records, default=str).encode())
        t.updated_at = datetime.now().isoformat()
        return {'table':name, 'written':len(records)}
    def discover(self, zone=None):
        results = []
        for z, data in self.zones.items():
            if zone and z != zone: continue
            for name, t in data['tables'].items(): results.append(t)
        return results
    def stats(self):
        return {z:{'tables':len(d['tables']),'records':sum(t.record_count for t in d['tables'].values()),'size_mb':round(sum(t.size_bytes for t in d['tables'].values())/1024/1024,2)} for z,d in self.zones.items()}

lake = DataLakeEngine("enterprise-lake")
lake.register_table('raw','raw_orders','json',{'order_id':'int','amount':'float'},['date'])
lake.register_table('raw','raw_clicks','json',{'session_id':'str','page':'str'},['date'])
lake.register_table('silver','clean_orders','parquet',{'order_id':'int','region':'str'},['date','region'])
lake.register_table('gold','daily_sales','parquet',{'date':'str','total':'float'},['date'])

for _ in range(10):
    lake.write_data('raw_orders', [{'order_id':random.randint(1,1000),'amount':round(random.uniform(10,5000),2)} for _ in range(random.randint(50,200))])
    lake.write_data('raw_clicks', [{'session_id':f'S_{random.randint(1,10000)}','page':f'/p/{random.randint(1,100)}'} for _ in range(random.randint(100,500))])
lake.write_data('clean_orders', [{'order_id':i,'region':random.choice(['华东','华南'])} for i in range(500)])
lake.write_data('daily_sales', [{'date':'2024-01-15','total':15000}])

print(f"数据湖: {lake.name}")
for zone, info in lake.stats().items():
    print(f"  {zone}: {info['tables']}表, {info['records']}条, {info['size_mb']}MB")
print("发现(raw):")
for t in lake.discover('raw'):
    print(f"  {t.name}({t.format}): {t.record_count}条")
print("✅ 验证通过 - 数据湖引擎运行正常")

📖 原理深入：现代数据存储架构

🏆 最佳实践：数据存储设计检查清单

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

📚 延伸阅读与参考

🤔 常见问题与解答