第13课：流批一体架构

⚡ 流批一体架构

流批一体架构旨在统一批处理和流处理，消除维护两套代码的负担。

🐍 Python实战：核心实战

# 流批一体框架
import time, random
from datetime import datetime
from collections import defaultdict

class BatchLayer:
    def __init__(self): self.data = []
    def ingest(self, records): self.data.extend(records)
    def compute(self, group_by, agg_field):
        groups = defaultdict(list)
        for r in self.data: groups[r.get(group_by,'unknown')].append(r.get(agg_field,0))
        return {k:round(sum(v),2) for k,v in groups.items()}

class SpeedLayer:
    def __init__(self): self.recent = []; self.window_size = 100
    def ingest(self, record):
        self.recent.append(record)
        if len(self.recent) > self.window_size: self.recent.pop(0)
    def compute(self, group_by, agg_field):
        groups = defaultdict(list)
        for r in self.recent: groups[r.get(group_by,'unknown')].append(r.get(agg_field,0))
        return {k:round(sum(v),2) for k,v in groups.items()}

class ServingLayer:
    def merge(self, batch_result, speed_result):
        all_keys = set(batch_result.keys()) | set(speed_result.keys())
        return {k: batch_result.get(k,0) + speed_result.get(k,0) for k in all_keys}

# Lambda架构测试
batch = BatchLayer()
speed = SpeedLayer()
serving = ServingLayer()

regions = ['华东','华南','华北']
# 批量层
batch.ingest([{'region':random.choice(regions),'amount':round(random.uniform(10,1000),2)} for _ in range(10000)])
# 速度层
for _ in range(50): speed.ingest({'region':random.choice(regions),'amount':round(random.uniform(10,1000),2)})

batch_result = batch.compute('region', 'amount')
speed_result = speed.compute('region', 'amount')
final = serving.merge(batch_result, speed_result)

print("Lambda架构:")
print(f"  批量层: {batch_result}")
print(f"  速度层: {speed_result}")
print(f"  合并: {final}")

# Kappa架构
class KappaEngine:
    def __init__(self): self.events = []
    def process(self, event):
        self.events.append(event)
    def query(self, group_by, agg_field, window=None):
        data = self.events[-window:] if window else self.events
        groups = defaultdict(list)
        for r in data: groups[r.get(group_by,'unknown')].append(r.get(agg_field,0))
        return {k:round(sum(v),2) for k,v in groups.items()}

kappa = KappaEngine()
for _ in range(10050): kappa.process({'region':random.choice(regions),'amount':round(random.uniform(10,1000),2)})
print(f"\nKappa架构: {kappa.query('region','amount')}")
print("✅ 验证通过 - 流批一体框架运行正常")

📖 原理深入：流处理核心理论

🏆 最佳实践：流处理设计检查清单

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

📚 延伸阅读与参考

🤔 常见问题与解答

🔑 关键技术选型决策

数据量级？
├── <1GB/天 → SQLite/PostgreSQL + Python脚本
├── 1-100GB/天 → 数据仓库(Snowflake/BQ) + dbt
├── 100GB-1TB/天 → Spark + 数据湖
└── >1TB/天 → Spark/Flink + 湖仓一体

延迟要求？
├── T+1 → 批处理(Spark) + Airflow
├── 小时级 → 微批(Spark Streaming) + Airflow
├── 分钟级 → 流处理(Flink) + Kafka
└── 秒级 → 纯流(Flink) + Kafka + 缓存

数据类型？
├── 纯结构化 → 数据仓库
├── 结构化+半结构化 → 湖仓一体
└── 全类型 → 数据湖 + 元数据管理