In this post we are going to cover partitions and throughput in Azure Event Hubs. These are 7 real Event Hubs interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: an IoT program will grow from one thousand to one hundred thousand devices this year, and you size Event Hubs for it.
A partition is an ordered slice of an event hub, and it is the unit of parallelism. Throughput units, processing units and capacity units are how much capacity the namespace has. Interviewers ask about this because both are hard to fix later: in Standard, the partition count can't change after creation, and a bad partition key or too few units shows up as lag and ServerBusy errors at peak.
Part 2 of the Real Time Azure Event Hubs Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the project
- Question 1: A teammate asks how many partitions a new event hub should have. How do you decide?
- Question 2: Events for one customer must stay in order. How do partition keys work?
- Question 3: One partition gets most of the traffic and its consumer falls behind. How do you find and fix a hot partition?
- Question 4: The producer gets a ServerBusy error at peak. What are throughput units and processing units?
- Question 5: Traffic doubles every holiday season. How do auto-inflate and scaling work?
- Question 6: Someone wants to add partitions to a live event hub. What happens to ordering and which tiers allow it?
- Scenario 7: An IoT program will grow from one thousand to one hundred thousand devices this year. How do you size Event Hubs?
- The full script
The short answer
Partitions set the maximum parallel consumers per consumer group; size them for peak, because Standard can't change them. A partition key keeps one customer's events together and in order. Capacity comes from throughput units (Basic, Standard), processing units (Premium) or capacity units (Dedicated); auto-inflate only scales up, and added partitions change the key mapping.
Let's set up the project
We run Microsoft's official Event Hubs emulator locally, with Azurite for the checkpoint store, and small Python scripts that use the azure-eventhub client, so every output is real. Features the emulator doesn't support, like auto-inflate and adding partitions, are shown as code. The emulator reads its entities from Config.json: orders with the consumer group billing, clicks, store-sales with the group dashboard, all with 4 partitions, plus orders-v2 with 8 and telemetry with 32. Before the first step we run python setup_storage.py once. It creates the checkpoint containers billing-checkpoints and dashboard-checkpoints in Azurite.
{
"UserConfig": {
"NamespaceConfig": [
{
"Type": "EventHub",
"Name": "emulatorNs1",
"Entities": [
{
"Name": "orders",
"PartitionCount": 4,
"ConsumerGroups": [
{
"Name": "billing"
}
]
},
{
"Name": "clicks",
"PartitionCount": 4,
"ConsumerGroups": []
},
{
"Name": "store-sales",
"PartitionCount": 4,
"ConsumerGroups": [
{
"Name": "dashboard"
}
]
},
{
"Name": "orders-v2",
"PartitionCount": 8,
"ConsumerGroups": []
},
{
"Name": "telemetry",
"PartitionCount": 32,
"ConsumerGroups": []
}
]
}
],
"LoggingConfig": {
"Type": "File"
}
}
}import os
from azure.storage.blob import BlobServiceClient
blob = BlobServiceClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"])
for name in ["billing-checkpoints", "dashboard-checkpoints"]:
blob.create_container(name) # checkpoint storesimport logging, os, threading, time
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import (
BlobCheckpointStore)
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "billing-checkpoints")
def start(mine):
client = EventHubConsumerClient.from_connection_string(
conn, consumer_group="billing", eventhub_name="orders",
checkpoint_store=store, load_balancing_interval=2)
threading.Thread(target=client.receive, kwargs=dict(
on_event=lambda p, event: None,
on_partition_initialize=lambda p: mine.add(p.partition_id),
on_partition_close=lambda p, _: mine.discard(p.partition_id),
)).start()
return client
owned = [set() for _ in range(6)] # 6 consumer instances
clients = [start(mine) for mine in owned]
time.sleep(30) # let them balance
print("6 consumers in one group, 4 partitions")
print("partitions per consumer:", sorted(len(m) for m in owned))
for client in clients:
client.close()import json, os
from azure.eventhub import EventData, EventHubProducerClient
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="orders")
events = ["cust-1 placed", "cust-2 placed", "cust-1 paid",
"cust-3 placed", "cust-2 paid", "cust-1 shipped",
"cust-3 paid", "cust-2 shipped", "cust-3 shipped"]
with producer:
for event in events: # arrival order
customer, status = event.split()
body = json.dumps({"customer": customer, "status": status})
producer.send_event(EventData(body), partition_key=customer)
print("sent", len(events), "order events, partition key = customer")import logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
consumer = EventHubConsumerClient.from_connection_string(
conn, consumer_group="$Default", eventhub_name="orders")
seen = []
def on_event(partition_context, event):
order = event.body_as_json()
pid = partition_context.partition_id
seen.append((pid, event.sequence_number,
order["customer"], order["status"]))
threading.Timer(6, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, starting_position="-1")
for pid, seq, customer, status in sorted(seen):
print(f"partition {pid} seq {seq} {customer} {status}")import os
from azure.eventhub import EventData, EventHubProducerClient
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="clicks")
with producer:
for n in range(8):
producer.send_event(EventData(f"click {n}")) # no key
counts = {}
for pid in producer.get_partition_ids():
p = producer.get_partition_properties(pid)
counts[pid] = p["last_enqueued_sequence_number"] + 1
print("8 events, no partition key")
print("per partition:", counts)import json, os, sys
from azure.eventhub import EventData, EventHubProducerClient
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
producer = EventHubProducerClient.from_connection_string(
conn, eventhub_name="store-sales")
per_store = {"store-7": 4, "store-11": 4, "store-15": 4,
"store-42": 4}
if sys.argv[1:] == ["flash-sale"]:
per_store = {"store-42": 16} # one store, a promo
with producer:
for store, count in per_store.items():
batch = producer.create_batch(partition_key=store)
for n in range(count):
sale = {"store": store, "sale": n}
batch.add(EventData(json.dumps(sale)))
producer.send_batch(batch)
print("sent", count, "sales for", store)import logging, os, threading
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import (
BlobCheckpointStore)
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "dashboard-checkpoints")
consumer = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="dashboard", eventhub_name="store-sales",
checkpoint_store=store)
done = []
def on_batch(partition_context, events):
if events:
done.extend(events) # update the dashboard
partition_context.update_checkpoint() # after the batch
threading.Timer(8, consumer.close).start()
with consumer:
consumer.receive_batch(on_event_batch=on_batch, max_wait_time=1,
starting_position="-1")
print("dashboard processed", len(done), "sales")import os
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import (
BlobCheckpointStore)
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "dashboard-checkpoints")
checkpoints = store.list_checkpoints("localhost", "store-sales",
"dashboard")
done = {c["partition_id"]: c["sequence_number"] for c in checkpoints}
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="dashboard", eventhub_name="store-sales")
with client:
for pid in client.get_partition_ids():
p = client.get_partition_properties(pid)
last = p["last_enqueued_sequence_number"]
lag = last - done.get(pid, -1)
print(f"partition {pid}: last enqueued {last:>2}, "
f"checkpoint {done.get(pid):>2}, lag {lag:>2}")import logging, os, threading
from azure.eventhub import (EventData, EventHubConsumerClient,
EventHubProducerClient)
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
keys = ["cust-1", "cust-2", "cust-3"]
def where_keys_land(hub):
producer = EventHubProducerClient.from_connection_string(
conn, eventhub_name=hub)
with producer:
for key in keys:
producer.send_event(EventData(key), partition_key=key)
count = len(producer.get_partition_ids())
consumer = EventHubConsumerClient.from_connection_string(
conn, consumer_group="$Default", eventhub_name=hub)
found = {}
def on_event(partition_context, event):
found[event.body_as_str()] = partition_context.partition_id
threading.Timer(5, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, starting_position="-1")
print(f"{hub}, {count} partitions:",
", ".join(f"{k} -> {found[k]}" for k in keys))
where_keys_land("orders") # 4 partitions
where_keys_land("orders-v2") # 8 partitionsimport os
from azure.eventhub import EventData, EventHubProducerClient
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
eventhub_name="telemetry")
with producer:
for d in range(1, 1001): # 1,000 devices
device = f"device-{d:04}"
producer.send_event(EventData('{"temp_c": 21.5}'),
partition_key=device)
counts = [producer.get_partition_properties(pid)
["last_enqueued_sequence_number"] + 1
for pid in producer.get_partition_ids()]
print(f"1000 devices, {len(counts)} partitions, "
f"events per partition: min {min(counts)}, max {max(counts)}")Question 1: A teammate asks how many partitions a new event hub should have. How do you decide?
The partition count is the maximum number of parallel consumers in one consumer group. So size it for peak throughput and for the readers you need.
First, the starting rule from the docs. In Standard, plan about 1 MB per second in and 2 MB per second out per partition. Divide the expected ingress and egress by those rates, take the larger result, and validate with a load test.
| Standard, per partition (starting point) | Example: expected | Partitions |
|---|---|---|
| ingress: about 1 MB/s | 4 MB/s in | 4 |
| egress: about 2 MB/s | 10 MB/s out, all consumer groups | 5 |
| take the larger, then load test | at least 5 |
Now let's see why the consumer count matters. instances.py starts six consumer instances in one consumer group on the 4-partition orders hub. They balance every 2 seconds instead of the default 30. We run python instances.py.
$ python instances.py
6 consumers in one group, 4 partitions
partitions per consumer: [0, 0, 1, 1, 1, 1]We get partitions per consumer: [0, 0, 1, 1, 1, 1]. Four instances own one partition each, and two own nothing: they sit idle.
The partition count doesn't change the price: we pay for throughput or processing units. But in Standard it's fixed at creation, and Premium and Dedicated can only add partitions, never remove them.
Size for next year's peak. Too many partitions adds processing complexity; too few, and consumers can't scale out.
Question 2: Events for one customer must stay in order. How do partition keys work?
A partition key is a string the sender sets, and Event Hubs hashes it to pick a partition. Events with the same key are stored together and delivered in the order they arrived.
Let's try it. orders.py sends nine order events for three customers, interleaved, with the customer as the key. Then read_orders.py reads them back sorted by partition and sequence. We run python orders.py and python read_orders.py.
$ python orders.py
sent 9 order events, partition key = customer
$ python read_orders.py
partition 0 seq 0 cust-3 placed
partition 0 seq 1 cust-3 paid
partition 0 seq 2 cust-3 shipped
partition 1 seq 0 cust-1 placed
partition 1 seq 1 cust-1 paid
partition 1 seq 2 cust-1 shipped
partition 3 seq 0 cust-2 placed
partition 3 seq 1 cust-2 paid
partition 3 seq 2 cust-2 shippedOn this emulator, cust-3 is on partition 0, cust-1 on partition 1, and cust-2 on partition 3. Each customer's placed, paid and shipped are in order, seq 0 to seq 2.
Order is guaranteed only within a partition. There's no global order across partitions, and that's fine: what matters is each customer's sequence. In Python, a whole batch lands on one partition, and a partition ID wins over a key.
Now without a key. We run python round_robin.py.
$ python round_robin.py
8 events, no partition key
per partition: {'0': 2, '1': 2, '2': 2, '3': 2}per partition: {'0': 2, '1': 2, '2': 2, '3': 2}: on the emulator, eight sends put two on each partition. No key is also better for availability. If a partition is unavailable, unkeyed events go to the others, while a keyed send gets an error.
Choose a key with many values, like the customer ID, never the country. And process each partition sequentially, or you lose the order you paid for.
Question 3: One partition gets most of the traffic and its consumer falls behind. How do you find and fix a hot partition?
Find it per partition. Here are the ways to see lag:
- Azure Monitor metrics - dimension EntityName: per event hub, no partition
- ConsumerLag - ApplicationMetricsLogs, Premium and Dedicated, per partition and group
- Compute it - last enqueued sequence number minus the checkpoint
Let's compute it. The store-sales hub is keyed by store. We run python sales.py, so four stores send four sales each. Then python dashboard.py processes them and checkpoints. Then python sales.py flash-sale: store-42 runs a promo.
$ python sales.py
sent 4 sales for store-7
sent 4 sales for store-11
sent 4 sales for store-15
sent 4 sales for store-42
$ python dashboard.py
dashboard processed 16 sales
$ python sales.py flash-sale
sent 16 sales for store-42The dashboard processed 16 sales, and then we get sent 16 sales for store-42, all with one key. Now the lag script, python lag.py.
$ python lag.py
partition 0: last enqueued 3, checkpoint 3, lag 0
partition 1: last enqueued 3, checkpoint 3, lag 0
partition 2: last enqueued 19, checkpoint 3, lag 16
partition 3: last enqueued 3, checkpoint 3, lag 0Partitions 0, 1 and 3 show lag 0. Partition 2, where store-42 lands on this emulator, shows last enqueued 19, checkpoint 3, lag 16.
Then the fixes:
- A key with more values - store + register, if order is per register
- No key - if order is not needed
- A faster consumer - for that partition: batching
More partitions won't help: one key still maps to one partition.
Alert on lag per partition, not per event hub. One hot key hides inside a healthy average.
Question 4: The producer gets a ServerBusy error at peak. What are throughput units and processing units?
ServerBusy means we hit our capacity, and ingress is throttled. Capacity is a different unit per tier.
| Unit | Tier | Capacity |
|---|---|---|
| Throughput unit (TU) | Basic, Standard | in: 1 MB/s or 1,000 events/s; out: 2 MB/s or 4,096 events/s; up to 40 |
| Processing unit (PU) | Premium | isolated CPU and memory: 1, 2, 4, 6, 8, 10, 12 or 16 per namespace |
| Capacity unit (CU) | Dedicated | single-tenant cluster: 1 to 10 self-serve, more by support request |
In Basic and Standard, it's throughput units. One unit allows 1 MB or 1,000 events per second in, whichever comes first, and 2 MB or 4,096 events per second out. We buy them per namespace, up to 40, shared by every event hub in it.
Over the ingress limit, senders get ServerBusy. Egress doesn't throw: readers are just held to the limit, with no error.
Premium uses processing units: isolated pods of CPU and memory, 1 to 16 per namespace, with no fixed per-unit limit. Dedicated uses capacity units on a single-tenant cluster.
The clients handle a busy server in different ways:
- Python SDK -
retry_total=3, backoff 0.8 s, exponential - Kafka - requests delayed; an error only past
request.timeout.ms
Raising the units is one command.
# Basic / Standard: throughput units, shared by the namespace
az eventhubs namespace update -g rg-shop -n shop-events \
--capacity 8Alert on the ThrottledRequests metric. Then raise the units, turn on auto-inflate, or move to Premium when 40 units aren't enough.
Question 5: Traffic doubles every holiday season. How do auto-inflate and scaling work?
In Standard, turn on auto-inflate. We set a maximum, and Event Hubs scales up the throughput units automatically as the load grows, so we avoid ServerBusy errors, up to that maximum.
The catch: auto-inflate never scales down by itself. After the holiday the units stay high, and we're billed by the units used each hour. So schedule a job to set the capacity back.
# before the season: raise the floor, cap auto-inflate
az eventhubs namespace update -g rg-shop -n shop-events \
--capacity 4 --enable-auto-inflate true \
--maximum-throughput-units 20
# after the season: auto-inflate never scales down
az eventhubs namespace update -g rg-shop -n shop-events \
--capacity 2The other tiers scale differently.
| Tier | How it scales |
|---|---|
| Standard | auto-inflate up to a maximum; you scale it down |
| Premium | processing units 1, 2, 4, 6, 8, 10, 12 or 16: you change them |
| Dedicated | self-serve cluster: 1 to 10 capacity units, out and in |
Premium has no auto-inflate: we choose the processing units and change them ourselves. Dedicated clusters are self-serve, out and in, in the portal.
And plan the partitions before the season, not during it. In Standard the count is fixed, and it caps how many consumers can read in parallel when the traffic doubles.
Raise the minimum ahead of the peak, set a sensible maximum, load test it, and scale back down when the season ends.
Question 6: Someone wants to add partitions to a live event hub. What happens to ordering and which tiers allow it?
Only Premium and Dedicated can add partitions to an existing event hub, and only up, never down. In Basic and Standard, the count is fixed at creation.
# Premium or Dedicated only: increase, never decrease
az eventhubs eventhub update -g rg-shop \
--namespace-name shop-events-premium -n orders \
--partition-count 8The ordering risk is key hashing. Existing data keeps its order, but for events sent after the change, the same key can map to a different partition. So a customer's new events may land somewhere else.
The emulator can't change a live event hub, so let's use a stand-in: the same three keys on the 4-partition orders hub and on orders-v2, with 8. We run python rehash.py.
$ python rehash.py
orders, 4 partitions: cust-1 -> 1, cust-2 -> 3, cust-3 -> 0
orders-v2, 8 partitions: cust-1 -> 2, cust-2 -> 7, cust-3 -> 0cust-1 moves from partition 1 to 2, and cust-2 from 3 to 7. cust-3 stays on 0.
Here is how each sender and reader reacts after partitions are added.
| Sender or reader | After partitions are added |
|---|---|
| Partition key | new events may go to another partition: ordering risk |
| Round robin (no key) | uses the new partitions within seconds |
| Partition ID | not affected; restart to see new partitions |
| Event processors | no automatic refresh: restart every instance |
| Kafka clients | metadata refresh, then a consumer group rebalance |
Drain first: let consumers finish every event in the existing partitions, then increase the count. Better still, size it right on day one.
Scenario 7: An IoT program will grow from one thousand to one hundred thousand devices this year. How do you size Event Hubs?
Start with written assumptions. Say each device sends a 1 KB reading every 10 seconds. At 100,000 devices, that's 10,000 events per second, about 10 MB per second in.
| At 100,000 devices (assumed: ~1 KB every 10 s) | Need | Sizing |
|---|---|---|
| Ingress | 10,000 events/s, ~10 MB/s | 10 TUs (1,000 events/s or 1 MB/s each) |
| Egress, 2 consumer groups | ~20 MB/s | 10 TUs (2 MB/s each) |
| Partitions | in 10 at 1 MB/s; out 20 at 2 MB/s | 10, with headroom: 32 |
| Connections | 100,000 devices | Standard: 5,000 per namespace |
First, throughput. A unit takes 1,000 events or 1 MB per second in, so ingress needs 10. Two consumer groups read 20 MB per second, at 2 per unit: also 10. That's the end state, so start low, with an auto-inflate maximum above 10.
Then partitions. 10 MB in at 1 per partition, and 20 out at 2, both give 10. Standard can't change it later, so we add headroom and take 32, the maximum. Let's check the spread with 1,000 device IDs as keys: python devices.py.
$ python devices.py
1000 devices, 32 partitions, events per partition: min 22, max 42On this emulator we get 1000 devices, 32 partitions, events per partition: min 22, max 42. Uneven, but spread.
The real limit is connections. Standard allows 5,000 brokered connections per namespace, far below 100,000 devices. Premium allows 10,000 per PU, and Dedicated 100,000 per CU. Microsoft recommends IoT Hub for device ingestion, or gateways that batch.
Devices into IoT Hub or gateways, then Event Hubs Standard with 32 partitions and auto-inflate, moving to Premium beyond 40 units.
Key points to remember
- Partitions = the maximum parallel consumers per consumer group; fixed in Standard
- A partition key keeps one customer together and in order; order is per partition
- Find a hot partition with per-partition lag: last enqueued minus checkpoint
- TUs (Basic, Standard), PUs (Premium), CUs (Dedicated); auto-inflate only scales up
- Adding partitions (Premium, Dedicated) changes the key mapping: drain first
The full script
This is a Python script that writes every file and runs every command above in order against the Event Hubs emulator: start the emulator with the Config.json above, run pip install azure-eventhub azure-eventhub-checkpointstoreblob azure-storage-blob confluent-kafka fastavro first, and set EVENTHUB_CONNECTION_STRING and STORAGE_CONNECTION_STRING.
Show the full script that writes the files and runs every step
# ========================================================================
# Event Hubs Interview Q&A #2 - Partitions and throughput
# TechBrothersIT. Names and sample data are invented for teaching.
# Every step below was run on the official Azure Event Hubs emulator (Docker image, version 2.2.1) with Azurite
# and azure-eventhub 5.15.1; the outputs in the video and the blog post are the real outputs.
# To run it yourself:
# 1. Start the emulator with Microsoft's installer (github.com/Azure/azure-event-hubs-emulator-installer),
# using a Config.json with the event hubs and consumer groups listed below.
# 2. pip install azure-eventhub azure-eventhub-checkpointstoreblob azure-storage-blob confluent-kafka fastavro
# 3. Set EVENTHUB_CONNECTION_STRING to the emulator connection string (Endpoint=sb://localhost;
# SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;)
# and STORAGE_CONNECTION_STRING to the Azurite blob connection string, then run python EventHubs_Interview_QA_02_Code.py.
# Entities in Config.json: orders (4 partitions, groups billing); clicks (4 partitions); store-sales (4 partitions, groups dashboard); orders-v2 (8 partitions); telemetry (32 partitions)
# Steps that restart the emulator say so in a comment: stop and start it by hand at that point.
# ========================================================================
import os, subprocess, time
def write(path, text):
if text is None:
if os.path.exists(path): os.remove(path)
return
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(text.strip("\n") + "\n")
def run(cmd, expect_error=False, allow_error=False, env=None):
print("\n$ " + cmd, flush=True)
rc = subprocess.run(cmd, shell=True, env=dict(os.environ, **(env or {}))).returncode
if rc != 0 and not (expect_error or allow_error):
raise SystemExit("stopped: this command was expected to succeed")
# Step 0: the starting files
write('setup_storage.py', """\
import os
from azure.storage.blob import BlobServiceClient
blob = BlobServiceClient.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"])
for name in ["billing-checkpoints", "dashboard-checkpoints"]:
blob.create_container(name) # checkpoint stores
""")
write('instances.py', """\
import logging, os, threading, time
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import (
BlobCheckpointStore)
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "billing-checkpoints")
def start(mine):
client = EventHubConsumerClient.from_connection_string(
conn, consumer_group="billing", eventhub_name="orders",
checkpoint_store=store, load_balancing_interval=2)
threading.Thread(target=client.receive, kwargs=dict(
on_event=lambda p, event: None,
on_partition_initialize=lambda p: mine.add(p.partition_id),
on_partition_close=lambda p, _: mine.discard(p.partition_id),
)).start()
return client
owned = [set() for _ in range(6)] # 6 consumer instances
clients = [start(mine) for mine in owned]
time.sleep(30) # let them balance
print("6 consumers in one group, 4 partitions")
print("partitions per consumer:", sorted(len(m) for m in owned))
for client in clients:
client.close()
""")
write('orders.py', """\
import json, os
from azure.eventhub import EventData, EventHubProducerClient
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="orders")
events = ["cust-1 placed", "cust-2 placed", "cust-1 paid",
"cust-3 placed", "cust-2 paid", "cust-1 shipped",
"cust-3 paid", "cust-2 shipped", "cust-3 shipped"]
with producer:
for event in events: # arrival order
customer, status = event.split()
body = json.dumps({"customer": customer, "status": status})
producer.send_event(EventData(body), partition_key=customer)
print("sent", len(events), "order events, partition key = customer")
""")
write('read_orders.py', """\
import logging, os, threading
from azure.eventhub import EventHubConsumerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
consumer = EventHubConsumerClient.from_connection_string(
conn, consumer_group="$Default", eventhub_name="orders")
seen = []
def on_event(partition_context, event):
order = event.body_as_json()
pid = partition_context.partition_id
seen.append((pid, event.sequence_number,
order["customer"], order["status"]))
threading.Timer(6, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, starting_position="-1")
for pid, seq, customer, status in sorted(seen):
print(f"partition {pid} seq {seq} {customer} {status}")
""")
write('round_robin.py', """\
import os
from azure.eventhub import EventData, EventHubProducerClient
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"], eventhub_name="clicks")
with producer:
for n in range(8):
producer.send_event(EventData(f"click {n}")) # no key
counts = {}
for pid in producer.get_partition_ids():
p = producer.get_partition_properties(pid)
counts[pid] = p["last_enqueued_sequence_number"] + 1
print("8 events, no partition key")
print("per partition:", counts)
""")
write('sales.py', """\
import json, os, sys
from azure.eventhub import EventData, EventHubProducerClient
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
producer = EventHubProducerClient.from_connection_string(
conn, eventhub_name="store-sales")
per_store = {"store-7": 4, "store-11": 4, "store-15": 4,
"store-42": 4}
if sys.argv[1:] == ["flash-sale"]:
per_store = {"store-42": 16} # one store, a promo
with producer:
for store, count in per_store.items():
batch = producer.create_batch(partition_key=store)
for n in range(count):
sale = {"store": store, "sale": n}
batch.add(EventData(json.dumps(sale)))
producer.send_batch(batch)
print("sent", count, "sales for", store)
""")
write('dashboard.py', """\
import logging, os, threading
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import (
BlobCheckpointStore)
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "dashboard-checkpoints")
consumer = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="dashboard", eventhub_name="store-sales",
checkpoint_store=store)
done = []
def on_batch(partition_context, events):
if events:
done.extend(events) # update the dashboard
partition_context.update_checkpoint() # after the batch
threading.Timer(8, consumer.close).start()
with consumer:
consumer.receive_batch(on_event_batch=on_batch, max_wait_time=1,
starting_position="-1")
print("dashboard processed", len(done), "sales")
""")
write('lag.py', """\
import os
from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import (
BlobCheckpointStore)
store = BlobCheckpointStore.from_connection_string(
os.environ["STORAGE_CONNECTION_STRING"], "dashboard-checkpoints")
checkpoints = store.list_checkpoints("localhost", "store-sales",
"dashboard")
done = {c["partition_id"]: c["sequence_number"] for c in checkpoints}
client = EventHubConsumerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
consumer_group="dashboard", eventhub_name="store-sales")
with client:
for pid in client.get_partition_ids():
p = client.get_partition_properties(pid)
last = p["last_enqueued_sequence_number"]
lag = last - done.get(pid, -1)
print(f"partition {pid}: last enqueued {last:>2}, "
f"checkpoint {done.get(pid):>2}, lag {lag:>2}")
""")
write('rehash.py', """\
import logging, os, threading
from azure.eventhub import (EventData, EventHubConsumerClient,
EventHubProducerClient)
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
conn = os.environ["EVENTHUB_CONNECTION_STRING"]
keys = ["cust-1", "cust-2", "cust-3"]
def where_keys_land(hub):
producer = EventHubProducerClient.from_connection_string(
conn, eventhub_name=hub)
with producer:
for key in keys:
producer.send_event(EventData(key), partition_key=key)
count = len(producer.get_partition_ids())
consumer = EventHubConsumerClient.from_connection_string(
conn, consumer_group="$Default", eventhub_name=hub)
found = {}
def on_event(partition_context, event):
found[event.body_as_str()] = partition_context.partition_id
threading.Timer(5, consumer.close).start()
with consumer:
consumer.receive(on_event=on_event, starting_position="-1")
print(f"{hub}, {count} partitions:",
", ".join(f"{k} -> {found[k]}" for k in keys))
where_keys_land("orders") # 4 partitions
where_keys_land("orders-v2") # 8 partitions
""")
write('devices.py', """\
import os
from azure.eventhub import EventData, EventHubProducerClient
producer = EventHubProducerClient.from_connection_string(
os.environ["EVENTHUB_CONNECTION_STRING"],
eventhub_name="telemetry")
with producer:
for d in range(1, 1001): # 1,000 devices
device = f"device-{d:04}"
producer.send_event(EventData('{"temp_c": 21.5}'),
partition_key=device)
counts = [producer.get_partition_properties(pid)
["last_enqueued_sequence_number"] + 1
for pid in producer.get_partition_ids()]
print(f"1000 devices, {len(counts)} partitions, "
f"events per partition: min {min(counts)}, max {max(counts)}")
""")
run('python setup_storage.py')
# Step 1: 6 consumer instances in one consumer group on a 4-partition event hub (load balancing every 2 s)
run('python instances.py')
# Step 2: 9 order events for 3 customers, partition key = customer, read back sorted by partition and sequence
run('python orders.py')
run('python read_orders.py')
# Step 3: 8 sends with no partition key: round robin, counted per partition
run('python round_robin.py')
# Step 4: 4 stores send 4 sales each; the dashboard processes and checkpoints; then store-42 runs a flash sale
run('python sales.py')
run('python dashboard.py')
run('python sales.py flash-sale')
# Step 5: Lag per partition: last enqueued sequence number minus the checkpoint
run('python lag.py')
# Step 6: The same keys on a 4-partition and an 8-partition event hub
run('python rehash.py')
# Step 7: 1,000 device IDs as partition keys on a 32-partition event hub
run('python devices.py')
Part 3: Producers and Sending Events
The sample data on this page is invented for teaching. Every terminal output is the real output of the command above it, run on the official Azure Event Hubs emulator (2.2.1) with Azurite and azure-eventhub 5.15.1. Features the emulator does not support are shown as code. Part of the Real Time Azure Event Hubs Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Azure Event Hubs documentation (learn.microsoft.com/azure/event-hubs) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.