Azure Event Hubs basics: create a namespace, send events and read them with Python

#The goal

You've been asked to look at a pipeline that uses Event Hubs, or to feed events into one. Event Hubs is a managed service for high-volume event streams. It's a lot like Kafka, and it even has a Kafka-compatible endpoint. Here, you'll build a small one, send events and read them back.

#The ideas you need

Consumers read events without removing them. Events stay until the retention period ends, so a new consumer group can read older data too.

#Step 1: create the resources

RG=rg-eventhub-lab
NS=ehns-lab-$RANDOM
EH=orders

az group create -n "$RG" -l centralindia

az eventhubs namespace create -g "$RG" -n "$NS" -l centralindia --sku Standard

az eventhubs eventhub create -g "$RG" --namespace-name "$NS" -n "$EH" \
  --partition-count 2 \
  --cleanup-policy Delete \
  --retention-time-in-hours 24

If your CLI rejects the retention flags, run az eventhubs eventhub create --help, as the names have changed across versions.

Choose the partition count with care, because on the Standard tier you can't change it later. For sizing, see tiers and auto-inflate.

#Step 2: grant access with Entra ID

Avoid connection strings where you can. Use a data role:

ME=$(az ad signed-in-user show --query id -o tsv)
SCOPE=$(az eventhubs namespace show -g "$RG" -n "$NS" --query id -o tsv)

az role assignment create --assignee-object-id "$ME" --assignee-principal-type User \
  --role "Azure Event Hubs Data Owner" --scope "$SCOPE"

Use Sender or Receiver for real applications. Role assignments can take a few minutes to apply.

#Step 3: send events

pip install azure-eventhub azure-identity

Save this as send.py. Replace the namespace name:

import json
from azure.eventhub import EventHubProducerClient, EventData
from azure.identity import DefaultAzureCredential

NAMESPACE = "ehns-lab-12345.servicebus.windows.net"
EVENT_HUB = "orders"

producer = EventHubProducerClient(
    fully_qualified_namespace=NAMESPACE,
    eventhub_name=EVENT_HUB,
    credential=DefaultAzureCredential(),
)

with producer:
    batch = producer.create_batch(partition_key="customer-100")
    for i in range(10):
        body = json.dumps({"order_id": i, "customer": "customer-100", "amount": 10 + i})
        batch.add(EventData(body))
    producer.send_batch(batch)

print("Sent 10 events")

DefaultAzureCredential picks up your az login session here, and a managed identity when running in Azure.

Events with the same partition_key go to the same partition, so they stay in order relative to each other. Ordering isn't guaranteed across partitions.

If you add more events than fit, batch.add raises ValueError. Real code sends the batch and starts a new one when that happens.

#Step 4: read events

Save this as receive.py:

from azure.eventhub import EventHubConsumerClient
from azure.identity import DefaultAzureCredential

NAMESPACE = "ehns-lab-12345.servicebus.windows.net"
EVENT_HUB = "orders"

def on_event(partition_context, event):
    print(partition_context.partition_id, event.sequence_number, event.body_as_str())

client = EventHubConsumerClient(
    fully_qualified_namespace=NAMESPACE,
    eventhub_name=EVENT_HUB,
    consumer_group="$Default",
    credential=DefaultAzureCredential(),
)

with client:
    # "-1" means start from the beginning of the stream
    client.receive(on_event=on_event, starting_position="-1")

Run it in one terminal, and then run send.py in another. Press Ctrl+C to stop.

Run it twice and you'll see the same events again. It has no memory of where it stopped, which brings us to checkpoints.

#Step 5: save your position

A checkpoint stores how far you've read, in a Blob Storage container:

pip install azure-eventhub-checkpointstoreblob

az storage account create -g "$RG" -n stehlab$RANDOM --sku Standard_LRS

Create a container, and grant yourself Storage Blob Data Contributor on the account. Then:

from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()

store = BlobCheckpointStore(
    blob_account_url="https://stehlab12345.blob.core.windows.net",
    container_name="checkpoints",
    credential=credential,
)

def on_event(partition_context, event):
    print(event.body_as_str())
    partition_context.update_checkpoint(event)

client = EventHubConsumerClient(
    fully_qualified_namespace="ehns-lab-12345.servicebus.windows.net",
    eventhub_name="orders",
    consumer_group="$Default",
    credential=credential,
    checkpoint_store=store,
)

with client:
    client.receive(on_event=on_event, starting_position="-1")

Now a restart continues from where it left off. Calling update_checkpoint for each event is slow at high volume, so real consumers do it every few seconds or every N events. That also means a crash can replay events, so make your processing able to handle duplicates.

The store also lets several copies of the consumer share partitions between them. If they keep stealing partitions from each other, see epoch and ownership errors.

#Step 6: check its health

ID=$(az eventhubs namespace show -g "$RG" -n "$NS" --query id -o tsv)

az monitor metrics list --resource "$ID" \
  --metric IncomingMessages OutgoingMessages ThrottledRequests \
  --interval PT5M --aggregation Total -o table

#Common errors

#Clean up

az group delete -n "$RG" --yes --no-wait

Caution

This deletes the namespace, its events and your checkpoint storage account. Confirm that the group holds only lab resources before you run it.

#Prevent it next time

• Use one consumer group per application, so readers don't affect each other.
• Choose partition keys that spread load evenly. A single busy key can overload a partition.
• Alert on throttling, and on consumers falling behind.
• If your apps already speak Kafka, see the Kafka endpoint errors post, and choosing a streaming platform.