第14课：窗口与水位线

⏱️ 窗口与水位线

窗口将无界流切分为有界切片进行分析，水位线处理乱序事件和迟到数据。

🐍 Python实战：核心实战

# 窗口计算引擎
import time, random
from datetime import datetime, timedelta
from collections import defaultdict

class Event:
    def __init__(self, key, value, event_time, process_time=None):
        self.key = key; self.value = value; self.event_time = event_time; self.process_time = process_time or datetime.now()

class WindowManager:
    def __init__(self):
        self.windows = defaultdict(list)
    def tumbling(self, events, window_size_sec):
        windows = defaultdict(list)
        for e in events:
            wid = int(e.event_time.timestamp() / window_size_sec)
            windows[wid].append(e)
        return {wid: {'events':len(evts),'sum':round(sum(e.value for e in evts),2),'avg':round(sum(e.value for e in evts)/len(evts),2)} for wid, evts in windows.items()}
    def sliding(self, events, window_size_sec, slide_sec):
        windows = defaultdict(list)
        for e in events:
            start = int(e.event_time.timestamp() / slide_sec)
            for wid in range(start, start + window_size_sec // slide_sec + 1):
                windows[wid * slide_sec].append(e)
        return {wid: {'count':len(evts),'avg':round(sum(e.value for e in evts)/len(evts),2)} for wid, evts in sorted(windows.items())[:5]}

class Watermark:
    def __init__(self, allowed_lateness_sec=5):
        self.allowed_lateness = allowed_lateness_sec; self.current_watermark = None
    def update(self, event):
        wm = event.event_time - timedelta(seconds=self.allowed_lateness)
        if self.current_watermark is None or wm > self.current_watermark:
            self.current_watermark = wm
    def is_late(self, event):
        return event.event_time < self.current_watermark if self.current_watermark else False

# 生成测试事件(含乱序)
events = []
base = datetime(2024, 1, 1, 0, 0, 0)
for i in range(100):
    # 80%按序, 20%乱序
    if random.random() < 0.8:
        et = base + timedelta(seconds=i)
    else:
        et = base + timedelta(seconds=max(0, i - random.randint(1, 10)))
    events.append(Event(f"sensor_{random.randint(1,5)}", round(random.uniform(10,50),2), et))

# 按处理时间排序(模拟乱序到达)
random.shuffle(events)

wm = WindowManager()
watermark = Watermark(5)

late_count = 0
for e in events:
    watermark.update(e)
    if watermark.is_late(e): late_count += 1

tumbling = wm.tumbling(sorted(events, key=lambda e: e.event_time), 10)
sliding = wm.sliding(sorted(events, key=lambda e: e.event_time), 30, 10)

print("窗口计算引擎:")
print(f"  事件数: {len(events)}, 迟到: {late_count}")
print(f"  水位线: {watermark.current_watermark}")
print(f"  滚动窗口(10s): {len(tumbling)}个窗口")
for wid, stats in sorted(tumbling.items())[:3]:
    print(f"    窗口{wid}: {stats}")
print(f"  滑动窗口(30s/10s): {len(sliding)}个窗口")
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 + 缓存

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