Forge Raw Data into
Actionable Intelligence

1from kafka import KafkaProducer
2import json
3from datetime import datetime
4
5# Configure Kafka producer
6producer = KafkaProducer(
7    bootstrap_servers=['kafka-1:9092', 'kafka-2:9092'],
8    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
9    acks='all',
10    compression_type='lz4'
11)
12
13# Send telemetry event
14def send_telemetry(sensor_id, metric_name, value):
15    event = {
16        'event_id': f"{sensor_id}-{int(datetime.now().timestamp()*1000)}",
17        'timestamp': int(datetime.now().timestamp() * 1000),
18        'source_type': 'iot_sensor',
19        'source_id': sensor_id,
20        'metric_name': metric_name,
21        'metric_value': value,
22        'tags': {'facility': 'plant-A'}
23    }
24    
25    producer.send('raw-telemetry', value=event)
26    producer.flush()
27    return event
28
29# Example usage
30event = send_telemetry('sensor-001', 'temperature', 75.5)
31print(f"Sent event: {event['event_id']}")

1# Create high-throughput telemetry topic
2kafka-topics.sh --create \
3  --bootstrap-server kafka-1:9092 \
4  --topic raw-telemetry \
5  --partitions 12 \
6  --replication-factor 3 \
7  --config retention.ms=604800000 \
8  --config compression.type=lz4 \
9  --config min.insync.replicas=2
10
11# Verify topic configuration
12kafka-topics.sh --describe \
13  --bootstrap-server kafka-1:9092 \
14  --topic raw-telemetry
15
16# Monitor consumer lag
17kafka-consumer-groups.sh --describe \
18  --bootstrap-server kafka-1:9092 \
19  --group streaming-processor

1from pyspark.sql import SparkSession
2from pyspark.sql.functions import *
3from pyspark.sql.types import *
4
5# Initialize Spark session
6spark = SparkSession.builder \
7    .appName("TelemetryAggregator") \
8    .config("spark.sql.shuffle.partitions", 12) \
9    .getOrCreate()
10
11# Read from Kafka
12raw_stream = spark \
13    .readStream \
14    .format("kafka") \
15    .option("kafka.bootstrap.servers", "kafka-1:9092") \
16    .option("subscribe", "raw-telemetry") \
17    .load()
18
19# Parse and aggregate
20parsed = raw_stream \
21    .select(from_json(col("value").cast("string"), schema).alias("data")) \
22    .select("data.*") \
23    .withWatermark("timestamp", "10 minutes")
24
25aggregated = parsed \
26    .groupBy(
27        window(col("timestamp"), "1 minute"),
28        col("source_type"),
29        col("metric_name")
30    ) \
31    .agg(
32        count("*").alias("count"),
33        avg("metric_value").alias("avg_value"),
34        expr("percentile_approx(metric_value, 0.95)").alias("p95")
35    )
36
37# Write to PostgreSQL
38query = aggregated \
39    .writeStream \
40    .foreachBatch(write_to_postgres) \
41    .outputMode("append") \
42    .option("checkpointLocation", "/checkpoints") \
43    .start()

1from fastapi import FastAPI, Query
2from datetime import datetime
3from typing import List
4import asyncpg
5
6app = FastAPI(title="StreamForge API")
7
8@app.get("/api/v1/metrics")
9async def get_metrics(
10    source_type: str = Query(...),
11    metric_name: str = Query(...),
12    start_time: datetime = Query(...),
13    end_time: datetime = Query(...)
14):
15    """Retrieve aggregated metrics for a time range"""
16    
17    pool = app.state.db_pool
18    
19    query = """
20        SELECT 
21            window_start as timestamp,
22            avg_value as value,
23            count,
24            p95_value
25        FROM aggregations_1m
26        WHERE source_type = $1 
27          AND metric_name = $2
28          AND window_start >= $3
29          AND window_start <= $4
30        ORDER BY window_start
31    """
32    
33    async with pool.acquire() as conn:
34        rows = await conn.fetch(
35            query, source_type, metric_name, 
36            start_time, end_time
37        )
38    
39    return [dict(row) for row in rows]
40
41@app.get("/api/v1/anomalies")
42async def get_anomalies(severity: str = None):
43    """Retrieve detected anomalies"""
44    # Implementation here
45    pass