End-to-End Streaming Data Pipeline¶

"""
IoT Device Telemetry Simulator
Generates realistic telemetry data and sends to Event Hubs
"""

import asyncio
import json
import random
import os
from datetime import datetime, timedelta
from typing import List, Dict
from azure.eventhub.aio import EventHubProducerClient
from azure.eventhub import EventData

class IoTDeviceSimulator:
    """Simulates IoT device telemetry"""

    def __init__(self, device_id: str, location: str, device_type: str):
        self.device_id = device_id
        self.location = location
        self.device_type = device_type
        self.temperature = 25.0  # Base temperature
        self.humidity = 60.0     # Base humidity
        self.pressure = 1013.25  # Base pressure in hPa
        self.battery_level = 100.0
        self.signal_strength = -65.0

    def generate_telemetry(self) -> Dict:
        """Generate realistic telemetry data"""

        # Simulate gradual changes and occasional spikes
        self.temperature += random.uniform(-0.5, 0.5)
        self.humidity += random.uniform(-1.0, 1.0)
        self.pressure += random.uniform(-0.1, 0.1)
        self.battery_level -= random.uniform(0.01, 0.05)
        self.signal_strength += random.uniform(-2.0, 2.0)

        # Occasional anomalies (5% chance)
        if random.random() < 0.05:
            self.temperature += random.uniform(-20, 20)

        # Keep values in realistic ranges
        self.temperature = max(-10, min(50, self.temperature))
        self.humidity = max(0, min(100, self.humidity))
        self.pressure = max(950, min(1050, self.pressure))
        self.battery_level = max(0, min(100, self.battery_level))
        self.signal_strength = max(-100, min(-50, self.signal_strength))

        return {
            "DeviceId": self.device_id,
            "Location": self.location,
            "DeviceType": self.device_type,
            "Temperature": round(self.temperature, 2),
            "Humidity": round(self.humidity, 2),
            "Pressure": round(self.pressure, 2),
            "BatteryLevel": round(self.battery_level, 2),
            "SignalStrength": round(self.signal_strength, 2),
            "Timestamp": datetime.utcnow().isoformat(),
            "EventId": f"{self.device_id}-{datetime.utcnow().timestamp()}"
        }


class TelemetryProducer:
    """Sends telemetry to Event Hubs"""

    def __init__(self, connection_string: str, event_hub_name: str):
        self.connection_string = connection_string
        self.event_hub_name = event_hub_name
        self.devices: List[IoTDeviceSimulator] = []

    def create_devices(self, count: int = 10):
        """Create simulated IoT devices"""
        locations = ["Factory-A", "Factory-B", "Warehouse-1", "Warehouse-2", "Office"]
        device_types = ["Temperature", "Environmental", "HVAC", "Industrial"]

        for i in range(count):
            device = IoTDeviceSimulator(
                device_id=f"device-{i:03d}",
                location=random.choice(locations),
                device_type=random.choice(device_types)
            )
            self.devices.append(device)

        print(f"✅ Created {count} simulated devices")

    async def send_telemetry_batch(self, producer_client: EventHubProducerClient):
        """Send batch of telemetry from all devices"""
        try:
            event_data_batch = await producer_client.create_batch()

            for device in self.devices:
                telemetry = device.generate_telemetry()
                event = EventData(json.dumps(telemetry))
                event.properties = {
                    "DeviceType": device.device_type,
                    "Location": device.location
                }

                try:
                    event_data_batch.add(event)
                except ValueError:
                    # Batch full, send and create new one
                    await producer_client.send_batch(event_data_batch)
                    event_data_batch = await producer_client.create_batch()
                    event_data_batch.add(event)

            # Send remaining events
            if len(event_data_batch) > 0:
                await producer_client.send_batch(event_data_batch)

        except Exception as e:
            print(f"❌ Error sending batch: {str(e)}")

    async def run(self, duration_minutes: int = 60, interval_seconds: int = 5):
        """Run telemetry generation"""
        print(f"🚀 Starting telemetry generation...")
        print(f"   Duration: {duration_minutes} minutes")
        print(f"   Interval: {interval_seconds} seconds")
        print(f"   Devices: {len(self.devices)}")

        async with EventHubProducerClient.from_connection_string(
            conn_str=self.connection_string,
            eventhub_name=self.event_hub_name
        ) as producer:

            end_time = datetime.utcnow() + timedelta(minutes=duration_minutes)
            batch_count = 0

            while datetime.utcnow() < end_time:
                await self.send_telemetry_batch(producer)
                batch_count += 1

                if batch_count % 10 == 0:
                    print(f"📊 Sent {batch_count} batches ({batch_count * len(self.devices)} events)")

                await asyncio.sleep(interval_seconds)

        print(f"✅ Completed! Sent {batch_count} batches")


# Main execution
async def main():
    """Main execution function"""

    # Configuration
    connection_string = os.getenv("EVENT_HUB_CONNECTION_STRING")
    event_hub_name = os.getenv("EVENT_HUB_NAME", "device-telemetry")

    if not connection_string:
        raise ValueError("EVENT_HUB_CONNECTION_STRING environment variable not set")

    # Create and run producer
    producer = TelemetryProducer(connection_string, event_hub_name)
    producer.create_devices(count=20)

    # Run for 1 hour, sending every 5 seconds
    await producer.run(duration_minutes=60, interval_seconds=5)


if __name__ == "__main__":
    asyncio.run(main())

# =====================================================
# Delta Lake Processing in Synapse Spark
# =====================================================

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from delta.tables import DeltaTable

# Initialize Spark session
spark = SparkSession.builder \
    .appName("TelemetryDeltaProcessing") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# Storage configuration
storage_account = "<your-storage-account>"
container = "telemetry-data"
base_path = f"abfss://{container}@{storage_account}.dfs.core.windows.net"

# =====================================================
# 1. Create Delta Lake Tables
# =====================================================

# Create database
spark.sql("CREATE DATABASE IF NOT EXISTS telemetry")
spark.sql("USE telemetry")

# Create raw telemetry table
raw_table_path = f"{base_path}/delta/raw_telemetry"

spark.sql(f"""
CREATE TABLE IF NOT EXISTS raw_telemetry (
    EventId STRING,
    DeviceId STRING,
    Location STRING,
    DeviceType STRING,
    Temperature DOUBLE,
    Humidity DOUBLE,
    Pressure DOUBLE,
    BatteryLevel DOUBLE,
    SignalStrength DOUBLE,
    Timestamp TIMESTAMP,
    IngestionTime TIMESTAMP
)
USING DELTA
LOCATION '{raw_table_path}'
PARTITIONED BY (DATE(Timestamp))
""")

# Create aggregated metrics table
agg_table_path = f"{base_path}/delta/aggregated_metrics"

spark.sql(f"""
CREATE TABLE IF NOT EXISTS aggregated_metrics (
    DeviceId STRING,
    Location STRING,
    DeviceType STRING,
    WindowStart TIMESTAMP,
    WindowEnd TIMESTAMP,
    EventCount LONG,
    AvgTemperature DOUBLE,
    MinTemperature DOUBLE,
    MaxTemperature DOUBLE,
    AvgHumidity DOUBLE,
    AvgPressure DOUBLE
)
USING DELTA
LOCATION '{agg_table_path}'
PARTITIONED BY (DATE(WindowEnd))
""")

# =====================================================
# 2. Process Streaming Data
# =====================================================

# Read from Event Hub (or Synapse table)
# In this example, we'll read from the Synapse table written by Stream Analytics

raw_df = spark.table("raw_telemetry")

print("📊 Data Statistics:")
raw_df.select(
    count("*").alias("total_events"),
    countDistinct("DeviceId").alias("unique_devices"),
    min("Timestamp").alias("earliest_event"),
    max("Timestamp").alias("latest_event")
).show()

# =====================================================
# 3. Data Quality Checks
# =====================================================

# Identify anomalies
anomalies_df = raw_df.filter(
    (col("Temperature") > 80) | (col("Temperature") < -10) |
    (col("Humidity") > 100) | (col("Humidity") < 0) |
    (col("BatteryLevel") < 10)
).withColumn("AnomalyType",
    when(col("Temperature") > 80, "HighTemp")
    .when(col("Temperature") < -10, "LowTemp")
    .when(col("Humidity") > 100, "InvalidHumidity")
    .when(col("BatteryLevel") < 10, "LowBattery")
    .otherwise("Unknown")
)

print("\n⚠️  Detected Anomalies:")
anomalies_df.groupBy("AnomalyType", "DeviceId") \
    .count() \
    .orderBy(desc("count")) \
    .show()

# =====================================================
# 4. Optimize Delta Tables
# =====================================================

# Optimize raw telemetry table
print("\n🔧 Optimizing raw_telemetry table...")
spark.sql(f"OPTIMIZE telemetry.raw_telemetry ZORDER BY (DeviceId, Timestamp)")

# Vacuum old files (keep 7 days)
print("🧹 Running VACUUM...")
spark.sql(f"VACUUM telemetry.raw_telemetry RETAIN 168 HOURS")

# Update table statistics
print("📈 Updating statistics...")
spark.sql("ANALYZE TABLE telemetry.raw_telemetry COMPUTE STATISTICS")

# =====================================================
# 5. Create Aggregated Views
# =====================================================

# Device summary view
device_summary = raw_df.groupBy("DeviceId", "Location", "DeviceType") \
    .agg(
        count("*").alias("TotalEvents"),
        avg("Temperature").alias("AvgTemperature"),
        avg("Humidity").alias("AvgHumidity"),
        avg("BatteryLevel").alias("AvgBatteryLevel"),
        min("Timestamp").alias("FirstSeen"),
        max("Timestamp").alias("LastSeen")
    )

print("\n📊 Device Summary:")
device_summary.orderBy(desc("TotalEvents")).show(10)

# Save to Delta table
device_summary_path = f"{base_path}/delta/device_summary"
device_summary.write \
    .format("delta") \
    .mode("overwrite") \
    .save(device_summary_path)

print(f"✅ Delta Lake processing complete!")
print(f"   Raw telemetry: {raw_table_path}")
print(f"   Device summary: {device_summary_path}")