第11课：流处理概述

🌊 流处理概述

流处理是对无边界数据流进行实时处理的方法，与批处理的核心区别在于数据是持续到达的。

🐍 Python实战：核心实战

# 流处理引擎模拟
import time, random, json
from datetime import datetime
from collections import defaultdict
from dataclasses import dataclass, field
from threading import Thread
import queue

@dataclass
class StreamRecord:
    key: str; value: dict; timestamp: str; offset: int

class StreamSource:
    def __init__(self, name, rate=10):
        self.name = name; self.rate = rate; self.offset = 0; self.running = False
    def generate(self):
        self.offset += 1
        return StreamRecord(f"key_{random.randint(1,100)}",
            {'event': random.choice(['click','purchase','view','search']),
             'user_id': f"U{random.randint(1,1000)}", 'value': round(random.uniform(1,100),2)},
            datetime.now().isoformat(), self.offset)

class StreamOperator:
    def __init__(self, name, fn):
        self.name = name; self.fn = fn; self.processed = 0
    def process(self, record):
        self.processed += 1
        return self.fn(record)

class StreamSink:
    def __init__(self, name): self.name = name; self.output = []
    def write(self, record): self.output.append(record); return len(self.output)

class StreamingEngine:
    def __init__(self, name):
        self.name = name; self.operators = []; self.source = None; self.sink = None
        self.metrics = defaultdict(int)
    def set_source(self, source): self.source = source; return self
    def add_operator(self, op): self.operators.append(op); return self
    def set_sink(self, sink): self.sink = sink; return self
    def run(self, num_records=100):
        print(f"启动流: {self.name} ({num_records}条)")
        start = time.perf_counter()
        for _ in range(num_records):
            record = self.source.generate()
            for op in self.operators:
                record = op.process(record)
                if record is None: break
            if record and self.sink:
                self.sink.write(record)
        elapsed = (time.perf_counter()-start)*1000
        tps = num_records / (elapsed/1000) if elapsed > 0 else 0
        print(f"  处理: {num_records}条, {elapsed:.1f}ms, {tps:.0f}条/秒")
        for op in self.operators:
            print(f"  {op.name}: {op.processed}条")
        if self.sink:
            print(f"  {self.sink.name}: {len(self.sink.output)}条输出")
        return self.metrics

# 构建流处理管道
engine = StreamingEngine("用户行为流")
engine.set_source(StreamSource("events", 100))
engine.add_operator(StreamOperator("过滤", lambda r: r if r.value.get('event') != 'search' else None))
engine.add_operator(StreamOperator("增强", lambda r: StreamRecord(r.key, {**r.value, 'tier': 'VIP' if r.value['value'] > 50 else 'Normal'}, r.timestamp, r.offset)))
engine.set_sink(StreamSink("output"))
engine.run(500)

# 窗口聚合模拟
print("\n窗口聚合(5秒滚动窗口):")
window_size = 5
windows = defaultdict(list)
for i in range(50):
    ts = i * 0.1  # 0.1秒间隔
    window_id = int(ts / window_size)
    event = random.choice(['click','purchase','view'])
    windows[window_id].append({'ts':ts,'event':event,'value':round(random.uniform(1,50),2)})

for wid, events in sorted(windows.items()):
    total = sum(e['value'] for e in events)
    print(f"  窗口{wid}: {len(events)}事件, 总值{total:.1f}")

print("✅ 验证通过 - 流处理引擎运行正常")

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

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

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

📚 延伸阅读与参考

🤔 常见问题与解答