Azure Event Hubs

# Send events to Event Hubs using Python SDK
# pip install azure-eventhub

import asyncio
import json
from azure.eventhub.aio import EventHubProducerClient
from azure.eventhub import EventData
from datetime import datetime

CONNECTION_STRING = "Endpoint=sb://yournamespace.servicebus.windows.net/;..."
EVENT_HUB_NAME   = "sales-events"

async def send_events():
    producer = EventHubProducerClient.from_connection_string(
        conn_str=CONNECTION_STRING,
        eventhub_name=EVENT_HUB_NAME
    )

    async with producer:
        # Create a batch — Event Hubs sends in batches for efficiency
        event_batch = await producer.create_batch()

        # Add events to the batch
        for i in range(10):
            payload = {
                "event_id":   f"EVT-{i:04d}",
                "customer_id": f"CUST-{i * 100}",
                "amount":      round(99.99 + i * 10, 2),
                "timestamp":   datetime.utcnow().isoformat(),
                "event_type":  "purchase"
            }
            event_batch.add(EventData(json.dumps(payload)))

        # Send the whole batch in one network call
        await producer.send_batch(event_batch)
        print(f"Sent batch of {len(event_batch)} events")

asyncio.run(send_events())

# Read events from Event Hubs — consumer group pattern
# Multiple consumers can read the same stream independently

import asyncio
from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore

# Checkpoint store tracks which events have been processed
# If consumer restarts, it resumes from last checkpoint — not from the start
CHECKPOINT_STORE = BlobCheckpointStore.from_connection_string(
    conn_str="DefaultEndpointsProtocol=https;AccountName=yourstorage;...",
    container_name="eventhub-checkpoints"
)

async def on_event(partition_context, event):
    # Process the event
    payload = json.loads(event.body_as_str())
    print(f"Partition {partition_context.partition_id}: {payload['event_id']}")

    # Write to database, trigger downstream processing, etc.
    # ...

    # Checkpoint — marks this event as processed
    await partition_context.update_checkpoint(event)

async def consume():
    client = EventHubConsumerClient.from_connection_string(
        conn_str=CONNECTION_STRING,
        consumer_group="$Default",          # or a named consumer group
        eventhub_name=EVENT_HUB_NAME,
        checkpoint_store=CHECKPOINT_STORE,  # enables resume on restart
    )

    async with client:
        await client.receive(
            on_event=on_event,
            starting_position="-1",  # -1 = read from beginning of stream
        )

asyncio.run(consume())

// Event Hubs Capture — automatically archive events to ADLS Gen2
// Configure in Azure Portal: Event Hub → Features → Capture

// Capture writes Avro files to your storage account in this path format:
// {Namespace}/{EventHub}/{PartitionId}/{Year}/{Month}/{Day}/{Hour}/{Minute}/{Second}.avro

// Example file path:
// myNamespace/sales-events/0/2025/03/15/14/30/00.avro

// Read captured Avro files in Databricks:
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ReadEventHubCapture").getOrCreate()

# Read all captured Avro files for today
df = spark.read.format("avro").load(
    "abfss://capture@yourstorage.dfs.core.windows.net/myNamespace/sales-events/*/2025/03/15/*/*/*.avro"
)

# Event Hubs Capture wraps your payload in a Body column (binary)
import pyspark.sql.functions as F

df_parsed = df.withColumn(
    "payload",
    F.from_json(F.col("Body").cast("string"), schema)
)

df_parsed.select("payload.*").show()