Detect Industrial Equipment Anomalies in Real Time with Flink Agents and Apache Kafka

                      
                      import os
import json
from typing import Any, Dict
from kafka import KafkaConsumer, KafkaProducer
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# Configuration
KAFKA_BROKER = os.getenv('KAFKA_BROKER', 'localhost:9092')
INPUT_TOPIC = os.getenv('INPUT_TOPIC', 'equipment_data')
OUTPUT_TOPIC = os.getenv('OUTPUT_TOPIC', 'anomaly_alerts')

# Initialize Spark session
spark = SparkSession.builder \
    .appName('AnomalyDetection') \
    .getOrCreate()

# Initialize Kafka consumer and producer
consumer = KafkaConsumer(INPUT_TOPIC, bootstrap_servers=KAFKA_BROKER)
producer = KafkaProducer(bootstrap_servers=KAFKA_BROKER)

# Function to detect anomalies
def detect_anomalies(data: Dict[str, Any]) -> bool:
    # Simple anomaly detection logic (placeholder)
    return data['value'] > 100

# Main processing loop
try:
    for message in consumer:
        data = json.loads(message.value)
        if detect_anomalies(data):
            producer.send(OUTPUT_TOPIC, json.dumps(data).encode('utf-8'))
            print(f'Anomaly detected: {data}')
except Exception as e:
    print(f'Error: {str(e)}')
finally:
    consumer.close()
    producer.close()
    spark.stop()

if __name__ == '__main__':
    print('Starting anomaly detection service...')