Tutorial 05: Real-Time Streaming with Lambda Architecture¶

# save as: examples/streaming/earthquake_monitor.py
"""
Real-time earthquake event producer for Azure Event Hubs.
Polls the USGS real-time earthquake feed and sends new events
to Event Hubs as they appear.
"""
import asyncio
import aiohttp
import json
import hashlib
import time
from datetime import datetime, timezone
from azure.eventhub.aio import EventHubProducerClient
from azure.eventhub import EventData

# Configuration
USGS_FEED_URL = "https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/2.5_hour.geojson"
EVENT_HUB_CONNECTION_STR = "<YOUR_CONNECTION_STRING>"  # from Step 2d
EVENT_HUB_NAME = "earthquake-events"
POLL_INTERVAL_SECONDS = 30

# Track seen events to avoid duplicates
seen_events = set()

def get_partition_key(feature):
    """Assign partition key based on geographic region."""
    coords = feature["geometry"]["coordinates"]
    lon, lat = coords[0], coords[1]

    if lat > 50:
        return "alaska"
    elif lat > 18 and lat < 22 and lon > -162 and lon < -154:
        return "hawaii"
    elif lon < -115:
        return "us-west"
    elif lon < -90:
        return "us-central"
    else:
        return "us-east"

async def fetch_earthquakes(session):
    """Fetch latest earthquake data from USGS."""
    async with session.get(USGS_FEED_URL) as response:
        if response.status == 200:
            return await response.json()
        else:
            print(f"USGS API error: {response.status}")
            return None

async def send_events(producer, features):
    """Send new earthquake events to Event Hub."""
    new_count = 0
    async with producer:
        for feature in features:
            event_id = feature["properties"].get("code", "")
            if event_id in seen_events:
                continue

            seen_events.add(event_id)
            new_count += 1

            # Enrich the event
            event_data = {
                "event_id": event_id,
                "magnitude": feature["properties"].get("mag"),
                "place": feature["properties"].get("place"),
                "time": feature["properties"].get("time"),
                "longitude": feature["geometry"]["coordinates"][0],
                "latitude": feature["geometry"]["coordinates"][1],
                "depth_km": feature["geometry"]["coordinates"][2] if len(feature["geometry"]["coordinates"]) > 2 else None,
                "event_type": feature["properties"].get("type"),
                "tsunami_flag": feature["properties"].get("tsunami", 0),
                "alert_level": feature["properties"].get("alert"),
                "ingested_at": datetime.now(timezone.utc).isoformat(),
                "source": "usgs",
                "region": get_partition_key(feature),
            }

            # Create batch per partition key
            partition_key = get_partition_key(feature)
            event_batch = await producer.create_batch(partition_key=partition_key)
            event_batch.add(EventData(json.dumps(event_data)))
            await producer.send_batch(event_batch)

    return new_count

async def main():
    """Main polling loop."""
    print(f"Starting earthquake monitor...")
    print(f"Polling USGS feed every {POLL_INTERVAL_SECONDS}s")
    print(f"Sending to Event Hub: {EVENT_HUB_NAME}")
    print("-" * 50)

    producer = EventHubProducerClient.from_connection_string(
        conn_str=EVENT_HUB_CONNECTION_STR,
        eventhub_name=EVENT_HUB_NAME,
    )

    async with aiohttp.ClientSession() as http_session:
        while True:
            try:
                data = await fetch_earthquakes(http_session)
                if data and "features" in data:
                    new_count = await send_events(producer, data["features"])
                    total = len(data["features"])
                    ts = datetime.now().strftime("%H:%M:%S")
                    print(f"[{ts}] Polled {total} events, sent {new_count} new")
                else:
                    print(f"No data received from USGS")
            except Exception as e:
                print(f"Error: {e}")

            await asyncio.sleep(POLL_INTERVAL_SECONDS)

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

# Databricks notebook: notebooks/streaming/batch_reprocess.py
"""
Batch layer: reprocess all captured earthquake events from ADLS.
Computes accurate historical aggregations and writes to Gold layer.
"""
from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, count, avg, max as spark_max, min as spark_min, sum as spark_sum,
    from_json, window, to_timestamp, expr
)
from pyspark.sql.types import *

spark = SparkSession.builder.getOrCreate()

# Read all captured Avro files from Bronze
raw_df = spark.read.format("avro").load(
    "abfss://bronze@csadlsdev.dfs.core.windows.net/csa-eh-dev/earthquake-events/"
)

# Parse the JSON body from Event Hub capture
schema = StructType([
    StructField("event_id", StringType()),
    StructField("magnitude", DoubleType()),
    StructField("place", StringType()),
    StructField("time", LongType()),
    StructField("longitude", DoubleType()),
    StructField("latitude", DoubleType()),
    StructField("depth_km", DoubleType()),
    StructField("event_type", StringType()),
    StructField("tsunami_flag", IntegerType()),
    StructField("alert_level", StringType()),
    StructField("region", StringType()),
])

events_df = raw_df.select(
    from_json(col("Body").cast("string"), schema).alias("event")
).select("event.*")

events_df = events_df.withColumn(
    "event_time",
    to_timestamp(expr("timestamp_millis(time)"))
)

print(f"Total captured events: {events_df.count()}")
events_df.printSchema()

# Batch aggregation: hourly by region
hourly_agg = events_df.groupBy(
    window("event_time", "1 hour"),
    "region"
).agg(
    count("*").alias("event_count"),
    avg("magnitude").alias("avg_magnitude"),
    spark_max("magnitude").alias("max_magnitude"),
    spark_min("magnitude").alias("min_magnitude"),
    avg("depth_km").alias("avg_depth_km"),
    spark_sum(expr("CASE WHEN magnitude >= 5.0 THEN 1 ELSE 0 END")).alias("significant_count"),
)

hourly_agg = hourly_agg.select(
    col("window.start").alias("window_start"),
    col("window.end").alias("window_end"),
    "region", "event_count", "avg_magnitude", "max_magnitude",
    "min_magnitude", "avg_depth_km", "significant_count"
)

print(f"Hourly aggregation rows: {hourly_agg.count()}")
hourly_agg.show(10, truncate=False)

# Write batch results to Gold layer
hourly_agg.write.format("delta").mode("overwrite").save(
    "abfss://gold@csadlsdev.dfs.core.windows.net/streaming/earthquake_batch_hourly"
)

# Daily summary
daily_agg = events_df.groupBy(
    window("event_time", "1 day"),
    "region"
).agg(
    count("*").alias("event_count"),
    avg("magnitude").alias("avg_magnitude"),
    spark_max("magnitude").alias("max_magnitude"),
    spark_sum(expr("CASE WHEN tsunami_flag = 1 THEN 1 ELSE 0 END")).alias("tsunami_events"),
)

daily_agg = daily_agg.select(
    col("window.start").alias("date"),
    "region", "event_count", "avg_magnitude", "max_magnitude", "tsunami_events"
)

daily_agg.write.format("delta").mode("overwrite").save(
    "abfss://gold@csadlsdev.dfs.core.windows.net/streaming/earthquake_batch_daily"
)

print("Batch layer processing complete")

# save as: examples/streaming/weather_producer.py
"""
Weather data producer — sends weather observations to Event Hubs.
"""
import asyncio
import aiohttp
import json
from datetime import datetime, timezone
from azure.eventhub.aio import EventHubProducerClient
from azure.eventhub import EventData

WEATHER_API_URL = "https://api.weather.gov/alerts/active?area=CA"
EVENT_HUB_CONNECTION_STR = "<YOUR_WEATHER_EH_CONN_STR>"
EVENT_HUB_NAME = "weather-events"

async def fetch_and_send():
    producer = EventHubProducerClient.from_connection_string(
        conn_str=EVENT_HUB_CONNECTION_STR,
        eventhub_name=EVENT_HUB_NAME,
    )

    async with aiohttp.ClientSession() as session:
        async with session.get(WEATHER_API_URL) as response:
            data = await response.json()

    async with producer:
        batch = await producer.create_batch()
        for feature in data.get("features", []):
            event = {
                "alert_id": feature["properties"].get("id"),
                "event": feature["properties"].get("event"),
                "severity": feature["properties"].get("severity"),
                "headline": feature["properties"].get("headline"),
                "area": feature["properties"].get("areaDesc"),
                "onset": feature["properties"].get("onset"),
                "expires": feature["properties"].get("expires"),
                "ingested_at": datetime.now(timezone.utc).isoformat(),
            }
            try:
                batch.add(EventData(json.dumps(event)))
            except ValueError:
                await producer.send_batch(batch)
                batch = await producer.create_batch()
                batch.add(EventData(json.dumps(event)))

        await producer.send_batch(batch)
        print(f"Sent {len(data.get('features', []))} weather alerts")

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

# save as: examples/streaming/clickstream_simulator.py
"""
Simulated clickstream events for testing the streaming pipeline.
"""
import asyncio
import json
import random
import uuid
from datetime import datetime, timezone
from azure.eventhub.aio import EventHubProducerClient
from azure.eventhub import EventData

EVENT_HUB_CONNECTION_STR = "<YOUR_CLICKSTREAM_EH_CONN_STR>"
EVENT_HUB_NAME = "clickstream-events"

PAGES = ["/map", "/dashboard", "/reports", "/alerts", "/settings", "/api/earthquakes"]
ACTIONS = ["view", "click", "search", "filter", "export", "share"]

async def generate_events(producer, count=100):
    async with producer:
        batch = await producer.create_batch()
        for _ in range(count):
            event = {
                "session_id": str(uuid.uuid4()),
                "user_id": f"user-{random.randint(1, 50)}",
                "page": random.choice(PAGES),
                "action": random.choice(ACTIONS),
                "duration_ms": random.randint(100, 30000),
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }
            try:
                batch.add(EventData(json.dumps(event)))
            except ValueError:
                await producer.send_batch(batch)
                batch = await producer.create_batch()
                batch.add(EventData(json.dumps(event)))

        await producer.send_batch(batch)
        print(f"Sent {count} clickstream events")

async def main():
    producer = EventHubProducerClient.from_connection_string(
        conn_str=EVENT_HUB_CONNECTION_STR,
        eventhub_name=EVENT_HUB_NAME,
    )
    while True:
        await generate_events(producer, count=random.randint(10, 50))
        await asyncio.sleep(5)

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