pip install confluent-kafka pandas numpy datetime

# sensor_simulator.py
from confluent_kafka import Producer
import json
import time
import random
import logging

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Kafka配置
conf = {'bootstrap.servers': 'localhost:9092'}

# 創(chuàng)建Producer實(shí)例
producer = Producer(conf)

# 回調(diào)函數(shù)，確認(rèn)消息是否成功發(fā)送
def delivery_report(err, msg):
    if err is not None:
        logger.error(f'Message delivery failed: {err}')
    else:
        logger.info(f'Message delivered to {msg.topic()} [{msg.partition()}]')

# 模擬的傳感器ID列表
sensor_ids = [f'sensor_{i}' for i in range(1, 4)]

try:
    while True:
        for sensor_id in sensor_ids:
            # 模擬溫度數(shù)據(jù)：基礎(chǔ)值20°C，加上隨機(jī)波動(dòng)
            base_temp = 20.0
            fluctuation = random.uniform(-2, 2)
            # 小概率模擬一個(gè)異常峰值
            if random.random() < 0.02:
                fluctuation += random.choice([15, -15])
                logger.warning(f"Simulating anomaly for {sensor_id}")
                
            temperature = base_temp + fluctuation
            
            # 構(gòu)造消息內(nèi)容：傳感器ID、溫度值、時(shí)間戳
            message = {
                'sensor_id': sensor_id,
                'temperature': round(temperature, 2),
                'timestamp': int(time.time() * 1000)  # 毫秒時(shí)間戳
            }
            
            # 將消息轉(zhuǎn)換為JSON字符串并發(fā)送到Kafka
            message_json = json.dumps(message)
            producer.produce(
                'sensor-readings-raw',
                key=sensor_id,  # 使用sensor_id作為key，確保同一傳感器的數(shù)據(jù)進(jìn)入同一分區(qū)
                value=message_json,
                callback=delivery_report
            )
            
            # 立即輪詢以觸發(fā)回調(diào)
            producer.poll(0)
        
        # 每秒發(fā)送一輪所有傳感器的數(shù)據(jù)
        time.sleep(1)

except KeyboardInterrupt:
    logger.info("Producer interrupted by user.")
finally:
    # 等待所有未完成的消息被發(fā)送
    producer.flush()

# stream_processor.py
from confluent_kafka import Consumer, KafkaError
import pandas as pd
import json
import time
from collections import defaultdict
import logging

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Kafka消費(fèi)者配置
consumer_conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'sensor-stream-processor',
    'auto.offset.reset': 'earliest'  # 如果沒有偏移量，從最早的消息開始讀
}

# 創(chuàng)建Consumer實(shí)例
consumer = Consumer(consumer_conf)
consumer.subscribe(['sensor-readings-raw'])

# 初始化一個(gè)字典來存儲(chǔ)未處理的數(shù)據(jù)
# 結(jié)構(gòu): {sensor_id: list_of_(timestamp, temperature)_tuples}
raw_data_store = defaultdict(list)
# 窗口大?。ê撩耄?
WINDOW_SIZE_MS = 60 * 1000  # 1分鐘
# 微批處理間隔（秒）
BATCH_INTERVAL = 10

def process_window(sensor_id, data_list):
    """
    處理一個(gè)傳感器的一個(gè)窗口內(nèi)的數(shù)據(jù)。
    計(jì)算平均值，并返回結(jié)果。
    """
    if not data_list:
        return None
        
    # 將數(shù)據(jù)列表轉(zhuǎn)換為Pandas DataFrame以便計(jì)算
    df = pd.DataFrame(data_list, columns=['timestamp', 'temperature'])
    
    # 計(jì)算統(tǒng)計(jì)量
    avg_temp = df['temperature'].mean()
    max_temp = df['temperature'].max()
    min_temp = df['temperature'].min()
    count = len(df)
    
    # 獲取窗口的時(shí)間范圍
    window_end = max(df['timestamp'])
    window_start = window_end - WINDOW_SIZE_MS
    
    result = {
        'sensor_id': sensor_id,
        'window_start': window_start,
        'window_end': window_end,
        'avg_temperature': round(avg_temp, 2),
        'max_temperature': round(max_temp, 2),
        'min_temperature': round(min_temp, 2),
        'count': count,
        'processing_time': int(time.time() * 1000)
    }
    
    logger.info(f"Processed window for {sensor_id}: {result}")
    return result

def check_anomaly(current_value, previous_value, threshold=5.0):
    """
    簡單的異常檢測：檢查當(dāng)前值與前一個(gè)值的變化是否超過閾值。
    在實(shí)際應(yīng)用中，可以使用更復(fù)雜的算法（如Z-score, Isolation Forest）。
    """
    if previous_value is None:
        return False
    return abs(current_value - previous_value) > threshold

# 用于記錄每個(gè)傳感器上一個(gè)已知的值（用于簡單異常檢測）
last_values = {}

try:
    last_batch_time = time.time()
    logger.info("Starting stream processing...")
    
    while True:
        # 等待消息，超時(shí)時(shí)間為1秒
        msg = consumer.poll(1.0)
        
        if msg is None:
            # 沒有收到消息
            pass
        elif msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                # End of partition event
                logger.info(f'Reached end of partition {msg.partition()}')
            else:
                logger.error(f'Consumer error: {msg.error()}')
        else:
            # 成功收到消息
            try:
                # 解析消息
                message_json = msg.value().decode('utf-8')
                data = json.loads(message_json)
                
                sensor_id = data['sensor_id']
                temperature = data['temperature']
                timestamp = data['timestamp']
                
                # 將數(shù)據(jù)存入臨時(shí)存儲(chǔ)
                raw_data_store[sensor_id].append((timestamp, temperature))
                
                # --- 簡單實(shí)時(shí)異常檢測（逐條檢測） ---
                previous_value = last_values.get(sensor_id)
                if check_anomaly(temperature, previous_value):
                    logger.warning(f"ANOMALY DETECTED! Sensor: {sensor_id}, "
                                  f"Current: {temperature}, Previous: {previous_value}")
                last_values[sensor_id] = temperature
                # ----------------------------------
                
            except (json.JSONDecodeError, KeyError, UnicodeDecodeError) as e:
                logger.error(f"Error processing message: {e}, raw value: {msg.value()}")
        
        # 檢查是否到達(dá)微批處理時(shí)間
        current_time = time.time()
        if current_time - last_batch_time >= BATCH_INTERVAL:
            logger.info(f"--- Starting micro-batch processing ---")
            results = []
            
            # 獲取當(dāng)前時(shí)間（毫秒），作為窗口的結(jié)束邊界
            now_ms = int(current_time * 1000)
            window_start_boundary = now_ms - WINDOW_SIZE_MS
            
            # 處理每個(gè)傳感器的數(shù)據(jù)
            for sensor_id, data_list in raw_data_store.items():
                # 過濾出在最近1分鐘窗口內(nèi)的數(shù)據(jù)
                window_data = [(ts, temp) for (ts, temp) in data_list if ts >= window_start_boundary]
                # 更新存儲(chǔ)，只保留窗口內(nèi)的數(shù)據(jù)（防止內(nèi)存無限增長）
                raw_data_store[sensor_id] = window_data
                
                # 如果窗口內(nèi)有數(shù)據(jù)，則進(jìn)行處理
                if window_data:
                    result = process_window(sensor_id, window_data)
                    if result:
                        results.append(result)
            
            # 在這里，可以將results寫入數(shù)據(jù)庫、另一個(gè)Kafka Topic或發(fā)布出去
            # 例如: write_to_database(results)
            # 或者: produce_to_kafka('sensor-readings-aggregated', results)
            logger.info(f"Micro-batch completed. Processed {len(results)} sensor windows.")
            
            # 重置批處理計(jì)時(shí)器
            last_batch_time = current_time
            
except KeyboardInterrupt:
    logger.info("Consumer interrupted by user.")
finally:
    # 關(guān)閉消費(fèi)者，釋放資源
    consumer.close()

# comprehensive_stream_processor.py
"""
一個(gè)完整的流處理示例：消費(fèi)Kafka中的傳感器數(shù)據(jù)，進(jìn)行窗口聚合計(jì)算和簡單異常檢測。
注意：這是一個(gè)示例，生產(chǎn)環(huán)境需考慮狀態(tài)持久化、容錯(cuò)、并行性等更多因素。
"""
from confluent_kafka import Consumer, Producer, KafkaError
import pandas as pd
import json
import time
from collections import defaultdict
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - [%(filename)s:%(lineno)d] - %(message)s'
)
logger = logging.getLogger('StreamProcessor')

# #################### 配置參數(shù) ####################
KAFKA_BOOTSTRAP_SERVERS = 'localhost:9092'
KAFKA_RAW_TOPIC = 'sensor-readings-raw'
KAFKA_AGG_TOPIC = 'sensor-readings-aggregated'
KAFKA_ALERT_TOPIC = 'sensor-alerts'
CONSUMER_GROUP_ID = 'sensor-stream-processor-v1'

WINDOW_SIZE_MS = 60 * 1000  # 聚合窗口大?。?分鐘
BATCH_PROCESSING_INTERVAL = 10  # 微批處理間隔：10秒
ANOMALY_THRESHOLD = 5.0  # 異常檢測閾值：溫度變化超過5°C

# #################### Kafka 客戶端配置 ####################
consumer_conf = {
    'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS,
    'group.id': CONSUMER_GROUP_ID,
    'auto.offset.reset': 'earliest',
    'enable.auto.commit': False  # 禁用自動(dòng)提交，改為手動(dòng)提交
}

producer_conf = {'bootstrap.servers': KAFKA_BOOTSTRAP_SERVERS}

# #################### 初始化客戶端 ####################
consumer = Consumer(consumer_conf)
producer = Producer(producer_conf)
consumer.subscribe([KAFKA_RAW_TOPIC])

# #################### 狀態(tài)存儲(chǔ) ####################
# 存儲(chǔ)原始數(shù)據(jù)：{sensor_id: [(timestamp_ms, temperature), ...]}
raw_data_store = defaultdict(list)
# 存儲(chǔ)上一個(gè)值，用于異常檢測：{sensor_id: last_temperature}
last_values = {}

# #################### 工具函數(shù) ####################
def delivery_report(err, msg):
    """ Producer消息發(fā)送回調(diào)函數(shù) """
    if err is not None:
        logger.error(f'Message delivery failed ({msg.topic()}): {err}')
    else:
        logger.debug(f'Message delivered to {msg.topic()} [{msg.partition()}]')

def produce_to_kafka(topic, key, value):
    """ 發(fā)送消息到Kafka """
    try:
        producer.produce(
            topic,
            key=key,
            value=json.dumps(value),
            callback=delivery_report
        )
        producer.poll(0)  # 輪詢以服務(wù)回調(diào)隊(duì)列
    except BufferError as e:
        logger.error(f'Producer buffer error: {e}')
        producer.poll(10)  # 等待一些空間

def process_window(sensor_id, data_list, window_end_ms):
    """
    處理一個(gè)傳感器的一個(gè)窗口內(nèi)的數(shù)據(jù)，計(jì)算聚合統(tǒng)計(jì)量。
    Args:
        sensor_id: 傳感器ID
        data_list: 窗口內(nèi)的數(shù)據(jù)列表，元素為(timestamp, temperature)
        window_end_ms: 窗口結(jié)束時(shí)間戳（毫秒）
    Returns:
        dict: 聚合結(jié)果字典
    """
    if not data_list:
        return None

    df = pd.DataFrame(data_list, columns=['timestamp', 'temperature'])
    window_start_ms = window_end_ms - WINDOW_SIZE_MS

    result = {
        'sensor_id': sensor_id,
        'window_start_utc': window_start_ms,
        'window_end_utc': window_end_ms,
        'avg_temperature': round(df['temperature'].mean(), 2),
        'max_temperature': round(df['temperature'].max(), 2),
        'min_temperature': round(df['temperature'].min(), 2),
        'count_readings': len(df),
        'processing_time_utc': int(time.time() * 1000)
    }
    logger.info(f"Aggregation complete for {sensor_id}: Avg={result['avg_temperature']}°C")
    return result

def check_anomaly(sensor_id, current_temp, threshold):
    """
    簡單異常檢測：檢查當(dāng)前溫度與上一次溫度的變化是否超過閾值。
    Returns:
        bool: 是否是異常
        float: 變化量
    """
    last_temp = last_values.get(sensor_id)
    if last_temp is None:
        return False, 0.0

    change = abs(current_temp - last_temp)
    return change > threshold, change

def cleanup_old_data(current_time_ms):
    """ 清理所有傳感器中超出當(dāng)前窗口的舊數(shù)據(jù)，防止內(nèi)存無限增長 """
    cutoff = current_time_ms - WINDOW_SIZE_MS
    for sensor_id in list(raw_data_store.keys()):
        # 只保留在時(shí)間窗口內(nèi)的數(shù)據(jù)點(diǎn)
        raw_data_store[sensor_id] = [
            (ts, temp) for (ts, temp) in raw_data_store[sensor_id] if ts >= cutoff
        ]
        # 如果某個(gè)傳感器的數(shù)據(jù)列表為空，可選擇刪除該鍵以節(jié)省空間
        if not raw_data_store[sensor_id]:
            del raw_data_store[sensor_id]

# #################### 主處理循環(huán) ####################
def main():
    logger.info("Starting Kafka Stream Processing Application...")
    last_batch_time = time.time()
    running = True

    try:
        while running:
            # 1. 輪詢Kafka獲取新消息
            msg = consumer.poll(1.0)  # 超時(shí)時(shí)間1秒

            if msg is None:
                continue
            if msg.error():
                if msg.error().code() == KafkaError._PARTITION_EOF:
                    logger.debug('Reached end of partition')
                else:
                    logger.error(f'Consumer error: {msg.error()}')
                continue

            # 2. 處理單條消息
            try:
                message_value = msg.value().decode('utf-8')
                data = json.loads(message_value)

                sensor_id = data['sensor_id']
                temperature = data['temperature']
                timestamp_ms = data['timestamp']  # 假設(shè)數(shù)據(jù)中自帶事件時(shí)間戳

                # 2.1 將數(shù)據(jù)存入窗口緩存
                raw_data_store[sensor_id].append((timestamp_ms, temperature))

                # 2.2 實(shí)時(shí)異常檢測（逐條處理）
                is_anomaly, delta = check_anomaly(sensor_id, temperature, ANOMALY_THRESHOLD)
                if is_anomaly:
                    alert_message = {
                        'sensor_id': sensor_id,
                        'current_temperature': temperature,
                        'previous_temperature': last_values[sensor_id],
                        'delta': round(delta, 2),
                        'threshold': ANOMALY_THRESHOLD,
                        'timestamp': timestamp_ms,
                        'alert_time': int(time.time() * 1000)
                    }
                    logger.warning(f"ANOMALY ALERT: {alert_message}")
                    # 將警報(bào)發(fā)送到另一個(gè)Kafka Topic
                    produce_to_kafka(KAFKA_ALERT_TOPIC, sensor_id, alert_message)

                # 更新上一個(gè)值的狀態(tài)
                last_values[sensor_id] = temperature

            except (KeyError, json.JSONDecodeError, ValueError, UnicodeDecodeError) as e:
                logger.error(f"Failed to process message: {e}. Raw value: {msg.value()}")

            # 3. 微批處理：檢查是否到達(dá)處理間隔
            current_time = time.time()
            if current_time - last_batch_time >= BATCH_PROCESSING_INTERVAL:
                logger.info("--- Starting micro-batch window aggregation ---")
                batch_processing_time_ms = int(current_time * 1000)
                window_end_boundary = batch_processing_time_ms  # 以處理時(shí)間作為窗口結(jié)束

                # 3.1 清理舊數(shù)據(jù)
                cleanup_old_data(batch_processing_time_ms)

                aggregated_results = []
                # 3.2 處理每個(gè)傳感器的窗口
                for sensor_id, data_list in raw_data_store.items():
                    if data_list:  # 確保有數(shù)據(jù)
                        result = process_window(sensor_id, data_list, window_end_boundary)
                        if result:
                            aggregated_results.append(result)
                            # 將聚合結(jié)果發(fā)送到Kafka
                            produce_to_kafka(KAFKA_AGG_TOPIC, sensor_id, result)

                logger.info(f"Micro-batch completed. Aggregated {len(aggregated_results)} windows.")

                # 3.3 手動(dòng)提交偏移量！確保在成功處理一批消息后再提交。
                # 注意：這里是簡單提交，生產(chǎn)環(huán)境應(yīng)更謹(jǐn)慎，例如確保producer的消息也已發(fā)送。
                consumer.commit(async=False) # 同步提交，更安全
                logger.debug("Kafka consumer offsets committed.")

                last_batch_time = current_time

    except KeyboardInterrupt:
        logger.info("Shutdown signal received.")
        running = False
    except Exception as e:
        logger.exception(f"Unexpected error occurred: {e}")
        running = False
    finally:
        logger.info("Shutting down...")
        consumer.close()
        producer.flush()  # 確保所有Producer消息都已發(fā)送
        logger.info("Shutdown complete.")

if __name__ == '__main__':
    main()