进阶教程：实时数据分析与自动化决策系统

1. 实时数据流处理架构

class StreamProcessor:
    def __init__(self):
        self.window_size = 60  # 滑动窗口大小(秒)
        self.analytics_engine = AnalyticsEngine()  # 复用之前的分析引擎
        
    def process_kafka_stream(self, topic):
        """从Kafka主题消费实时数据流"""
        consumer = KafkaConsumer(
            topic,
            bootstrap_servers='localhost:9092',
            value_deserializer=lambda x: json.loads(x.decode('utf-8'))
        )
        
        window_buffer = []
        for message in consumer:
            window_buffer.append(message.value)
            # 滑动窗口处理
            if len(window_buffer) >= self.window_size:
                self._analyze_window(window_buffer