In this post we are going to cover Schema Registry and event formats 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: five teams publish to shared event hubs, changes keep breaking consumers, and we introduce schemas safely.
Event Hubs moves bytes. It never looks inside the body, so nothing stops a producer from changing the shape of an event. Azure Schema Registry turns that shape into a contract that is checked before a change ships. Interviewers ask about this because broken event formats are one of the most common problems in shared streaming platforms.
Part 7 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 producer renamed a field and consumers broke. How does Azure Schema Registry help?
- Question 2: A teammate asks whether to send JSON or Avro. What are the trade-offs?
- Question 3: The schema must change without breaking existing consumers. How do compatibility modes work?
- Question 4: Someone asks how a consumer knows which schema an event was written with. How does it work?
- Question 5: Consumers need to tell event types apart in one event hub. How do you design event metadata?
- Question 6: Your manager asks who owns an event contract between teams. How do you manage it?
- Scenario 7: Five teams publish to shared event hubs and changes keep breaking consumers. How do you introduce schemas safely?
- The full script
The short answer
Azure Schema Registry turns the event format into a contract: schemas live in schema groups with a compatibility mode, producers send Avro with the schema ID in the content type, and consumers decode with the writer's schema. Add optional fields with defaults, route on event properties, and let the producer team own the contract with a check in every pull request.
Let's set up the project
We run Microsoft's official Event Hubs emulator locally, with small Python scripts that use the azure-eventhub client and fastavro, so every output is real. The emulator has no Schema Registry, so the registry parts are shown as code. Config.json creates three event hubs: orders-json, orders and shop-events. ehio.py holds small send and read helpers, avro_io.py encodes and decodes Avro, and the schemas folder has two versions of the OrderPlaced schema. Version 2 adds an optional coupon field.
{
"UserConfig": {
"NamespaceConfig": [
{
"Type": "EventHub",
"Name": "emulatorNs1",
"Entities": [
{
"Name": "orders-json",
"PartitionCount": 2,
"ConsumerGroups": [
{
"Name": "billing"
}
]
},
{
"Name": "orders",
"PartitionCount": 2,
"ConsumerGroups": [
{
"Name": "billing"
},
{
"Name": "analytics"
}
]
},
{
"Name": "shop-events",
"PartitionCount": 2,
"ConsumerGroups": [
{
"Name": "router"
}
]
}
]
}
],
"LoggingConfig": {
"Type": "File"
}
}
}import logging, os, threading
from azure.eventhub import EventHubConsumerClient, EventHubProducerClient
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
CONN = os.environ["EVENTHUB_CONNECTION_STRING"]
def send(hub, events, partition_id="0"):
"""Send a list of EventData as one batch to one partition."""
with EventHubProducerClient.from_connection_string(
CONN, eventhub_name=hub) as producer:
batch = producer.create_batch(partition_id=partition_id)
for event in events:
batch.add(event)
producer.send_batch(batch)
def read(hub, group, count, timeout=30):
"""Read `count` events from the start, in partition order."""
client = EventHubConsumerClient.from_connection_string(
CONN, consumer_group=group, eventhub_name=hub)
got, lock = [], threading.Lock()
def on_event(pc, event):
with lock:
seq = event.sequence_number
got.append((pc.partition_id, seq, event))
if len(got) == count:
threading.Thread(target=client.close).start()
timer = threading.Timer(timeout, client.close)
timer.start()
client.receive(on_event, starting_position="-1")
timer.cancel()
return [event for _, _, event in sorted(got, key=lambda g: g[:2])]
def body_bytes(event):
return b"".join(event.body)import io
import fastavro
def schema(version):
return fastavro.schema.load_schema(f"schemas/order_{version}.avsc")
def encode(record, version):
buf = io.BytesIO()
fastavro.schemaless_writer(buf, schema(version), record)
return buf.getvalue() # Avro binary, no header
def decode(data, writer_version, reader_version):
return fastavro.schemaless_reader(io.BytesIO(data),
schema(writer_version),
schema(reader_version)){"type": "record", "name": "OrderPlaced", "namespace": "shop",
"fields": [
{"name": "order_id", "type": "long"},
{"name": "customer_id", "type": "string"},
{"name": "amount", "type": "double"},
{"name": "currency", "type": "string"}
]}{"type": "record", "name": "OrderPlaced", "namespace": "shop",
"fields": [
{"name": "order_id", "type": "long"},
{"name": "customer_id", "type": "string"},
{"name": "amount", "type": "double"},
{"name": "currency", "type": "string"},
{"name": "coupon", "type": ["null", "string"], "default": null}
]}Question 1: A producer renamed a field and consumers broke. How does Azure Schema Registry help?
It turns the event format into a contract that is checked before a change ships. Schema Registry stores schemas inside your Event Hubs namespace, in schema groups, and each group has a compatibility mode. It is on Standard, Premium and Dedicated, not Basic.
First, let's see the breakage without it. Our producer publishes order 1001 as JSON with python publish_order.py 1001.
import json, sys
from azure.eventhub import EventData
from ehio import send
order_id = int(sys.argv[1])
order = {"order_id": order_id, "customer_id": "cust-7",
"amount": 25.50, "currency": "USD"}
event = EventData(json.dumps(order))
event.content_type = "application/json"
send("orders-json", [event])
print("published order", order_id, "fields:", sorted(order))$ python publish_order.py 1001
published order 1001 fields: ['amount', 'currency', 'customer_id', 'order_id']The fields are ['amount', 'currency', 'customer_id', 'order_id']. Now a teammate renames amount to total and publishes order 1002. Then the billing consumer, which checks its required fields, reads both. We run python publish_order.py 1002 and python billing_consumer.py.
import json, sys
from azure.eventhub import EventData
from ehio import send
order_id = int(sys.argv[1])
order = {"order_id": order_id, "customer_id": "cust-7",
"total": 25.50, "currency": "USD"} # was "amount"
event = EventData(json.dumps(order))
event.content_type = "application/json"
send("orders-json", [event])
print("published order", order_id, "fields:", sorted(order))from ehio import read
REQUIRED = ["order_id", "customer_id", "amount", "currency"]
for event in read("orders-json", "billing", count=2):
order = event.body_as_json()
missing = [f for f in REQUIRED if f not in order]
if missing:
print("order", order["order_id"], "REJECTED: missing", missing)
continue
print("order", order["order_id"],
f"billed {order['amount']:.2f} {order['currency']}")$ python publish_order.py 1002
published order 1002 fields: ['currency', 'customer_id', 'order_id', 'total']
$ python billing_consumer.py
order 1001 billed 25.50 USD
order 1002 REJECTED: missing ['amount']Order 1001 is billed 25.50 USD. Order 1002 is REJECTED: missing ['amount']. Nothing stopped that change, so the consumer found out in production.
With the registry, it works like this:
- Producer - encodes Avro with schema v2
- Schema Registry - the
ordersgroup, in Backward mode - Consumers - never see the bad event
A rename is a deleted field plus a new required field. Backward only allows deleting fields and adding optional fields, so the registry rejects the new version before any consumer sees an event.
Event Hubs itself never validates the body; the encoder and the registry do. So a pipeline registers schemas after a compatibility check in the pull request, never the applications at runtime.
Question 2: A teammate asks whether to send JSON or Avro. What are the trade-offs?
JSON is readable and every tool understands it, but every event repeats its field names, and nothing enforces the shape. Avro is a compact binary format that is always written with a schema, and it supports schema evolution.
| JSON | Avro | |
|---|---|---|
| Format | text, field names in every event | binary, values only |
| Schema | optional, nothing enforces it | always written with a schema |
| Evolution | by convention | rules: defaults, Backward, Forward |
| Debugging | readable anywhere | needs the writer schema |
Let's send the same order, 2001, as JSON and as Avro, read both bodies back, and then compare 10,000 orders. We run python send_formats.py, python read_formats.py and python size_test.py.
import json
from azure.eventhub import EventData
from avro_io import encode
from ehio import send
order = {"order_id": 2001, "customer_id": "cust-7",
"amount": 25.50, "currency": "USD"}
as_json = EventData(json.dumps(order))
as_json.content_type = "application/json"
as_avro = EventData(encode(order, "v1")) # values only, no names
as_avro.content_type = "avro/binary"
as_avro.properties = {"schema_version": "v1"}
send("orders", [as_json, as_avro])
print("sent order 2001 twice: as JSON and as Avro")from ehio import read, body_bytes
for event in read("orders", "analytics", count=2):
body = body_bytes(event)
print(f"{event.content_type:17} {len(body):3} bytes {body[:28]}")import json
from avro_io import encode
orders = [{"order_id": 100000 + i, "customer_id": f"cust-{i % 500}",
"amount": round(10 + i * 0.37, 2), "currency": "USD"}
for i in range(10000)]
json_bytes = sum(len(json.dumps(o).encode()) for o in orders)
avro_bytes = sum(len(encode(o, "v1")) for o in orders)
print("10,000 orders as JSON:", json_bytes, "bytes")
print("10,000 orders as Avro:", avro_bytes, "bytes")
print(f"Avro is {avro_bytes / json_bytes:.0%} of the JSON size")$ python send_formats.py
sent order 2001 twice: as JSON and as Avro
$ python read_formats.py
application/json 78 bytes b'{"order_id": 2001, "customer'
avro/binary 21 bytes b'\xa2\x1f\x0ccust-7\x00\x00\x00\x00\x00\x809@\x06USD'
$ python size_test.py
10,000 orders as JSON: 843880 bytes
10,000 orders as Avro: 237800 bytes
Avro is 28% of the JSON sizeThe JSON body is 78 bytes. The Avro body is 21 bytes: just the values, no field names. Across 10,000 orders, JSON is 843880 bytes and Avro is 237800 bytes, so Avro is 28% of the JSON size.
Does that matter for throughput? On Standard, 1 throughput unit takes in 1 MB per second or 1,000 events per second, whichever comes first. So smaller events help with the megabyte limit, not the event count.
The cost: Avro bytes are unreadable without the writer's schema. We need a registry, and debugging needs tooling.
My rule:
- JSON - low volume and partner facing events
- Avro with the registry - high volume internal streams
The registry also supports JSON Schema, but schema evolution checks are Avro only.
Question 3: The schema must change without breaking existing consumers. How do compatibility modes work?
Each schema group has a mode, and the mode decides which changes are allowed.
| Mode | Guarantee | Allowed changes |
|---|---|---|
| Backward | new-schema consumer reads old data | delete fields, add optional fields |
| Forward | old-schema consumer reads new data | add fields, delete optional fields |
| None | no compatibility checks | anything |
Let's try it. Version 2 adds a coupon field: optional, with a default of null. The upgraded producer sends order 2002 with the coupon FALL10. Then an old consumer on version 1, in the billing group, and a new consumer on version 2, in the analytics group, read both events. We run python publish_v2.py, python reader.py v1 billing and python reader.py v2 analytics.
from azure.eventhub import EventData
from avro_io import encode
from ehio import send
order = {"order_id": 2002, "customer_id": "cust-9", "amount": 40.00,
"currency": "USD", "coupon": "FALL10"} # new in v2
event = EventData(encode(order, "v2"))
event.content_type = "avro/binary"
event.properties = {"schema_version": "v2"}
send("orders", [event])
print("producer upgraded: sent order 2002 with schema v2")import sys
from avro_io import decode
from ehio import read, body_bytes
reader_version = sys.argv[1] # the schema this app knows
print("consumer with reader schema", reader_version)
for event in read("orders", sys.argv[2], count=3):
if event.content_type != "avro/binary":
continue # the JSON copy
writer = event.properties[b"schema_version"].decode()
order = decode(body_bytes(event), writer, reader_version)
print(" ", writer, order)$ python publish_v2.py
producer upgraded: sent order 2002 with schema v2
$ python reader.py v1 billing
consumer with reader schema v1
v1 {'order_id': 2001, 'customer_id': 'cust-7', 'amount': 25.5, 'currency': 'USD'}
v2 {'order_id': 2002, 'customer_id': 'cust-9', 'amount': 40.0, 'currency': 'USD'}
$ python reader.py v2 analytics
consumer with reader schema v2
v1 {'order_id': 2001, 'customer_id': 'cust-7', 'amount': 25.5, 'currency': 'USD', 'coupon': None}
v2 {'order_id': 2002, 'customer_id': 'cust-9', 'amount': 40.0, 'currency': 'USD', 'coupon': 'FALL10'}The old consumer reads both events. The version 2 event simply drops the coupon it doesn't know: order 2002 comes back as {'order_id': 2002, 'customer_id': 'cust-9', 'amount': 40.0, 'currency': 'USD'}.
The new consumer reads the old event too, and coupon gets its default, None. Order 2002 shows 'coupon': 'FALL10'.
So an optional field with a default passes both directions. The mode also tells us the deploy order:
- Backward - upgrade consumers first
- Forward - upgrade producers first
Every new field gets a default. A field without one is a breaking change waiting to happen.
Question 4: Someone asks how a consumer knows which schema an event was written with. How does it work?
Avro bytes carry no field names, so the reader needs the exact writer schema. It travels by reference: the event carries a schema ID, and the consumer gets that schema from the registry.
Let's see why it matters. We decode our two Avro events with guessed schemas, using python wrong_schema.py.
from avro_io import decode
from ehio import read, body_bytes
for event in read("orders", "$Default", count=3):
if event.content_type != "avro/binary":
continue
print("event properties:", event.properties)
for guess in ["v1", "v2"]: # pretend we don't know
try:
order = decode(body_bytes(event), guess, guess)
print(f" as {guess}: {order}")
except Exception as e:
print(f" as {guess}: {type(e).__name__}")$ python wrong_schema.py
event properties: {b'schema_version': b'v1'}
as v1: {'order_id': 2001, 'customer_id': 'cust-7', 'amount': 25.5, 'currency': 'USD'}
as v2: EOFError
event properties: {b'schema_version': b'v2'}
as v1: {'order_id': 2002, 'customer_id': 'cust-9', 'amount': 40.0, 'currency': 'USD'}
as v2: {'order_id': 2002, 'customer_id': 'cust-9', 'amount': 40.0, 'currency': 'USD', 'coupon':
'FALL10'}The version 1 event read as v2 fails with EOFError. The version 2 event read as v1 silently drops the coupon, with no error at all.
In our demo, a schema_version property names the writer schema. With Azure Schema Registry, the flow looks like this:
- Producer - writes the Avro body
- Event hub - the content type is
avro/binary+ID - Consumer - decodes with the writer schema
- Schema Registry - returns the schema by ID, and the consumer caches it
The Python Avro encoder does this for us: encode sets the body and the content type with the schema ID. On the consumer, decode reads the ID from the content type, fetches the schema once and caches it, so the registry isn't called for every event. We can pass our own reader schema too. The emulator has no registry, so this part is code only:
from azure.eventhub import EventData
from azure.identity import DefaultAzureCredential
from azure.schemaregistry import SchemaRegistryClient
from azure.schemaregistry.encoder.avroencoder import AvroEncoder
registry = SchemaRegistryClient(
"tbshop-ns.servicebus.windows.net", DefaultAzureCredential())
encoder = AvroEncoder(client=registry, group_name="orders")
# producer: Avro binary body + content type with the schema ID
event = encoder.encode(order, schema=ORDER_V2_AVSC,
message_type=EventData)
print(event.content_type) # avro/binary+<schema ID>
# consumer: schema fetched by ID once, then from a local cache
order = encoder.decode(event, readers_schema=ORDER_V1_AVSC)Decoding with your own schema as the writer schema. Always decode with the writer's schema, then resolve to the reader schema your app knows.
Question 5: Consumers need to tell event types apart in one event hub. How do you design event metadata?
Put the routing facts in event properties, not only in the body. Application properties are a small key value map next to the body, so a consumer can decide what to do without parsing the payload.
Here is the envelope we use:
event_type+event_version- route and pick the schemasource- which system sent itmessage_id- unique per event: the dedupe keycorrelation_id- follow one order across eventscontent_type- how to decode the body
Our shop API publishes six events into one hub, shop-events, including a brand new gift card type. Then the router reads only the properties. We run python publish_events.py and python router.py.
import json
from azure.eventhub import EventData
from ehio import send
EVENTS = [("evt-1", "OrderPlaced", "order-3001"),
("evt-2", "PaymentCaptured", "order-3001"),
("evt-3", "OrderPlaced", "order-3002"),
("evt-4", "OrderShipped", "order-3001"),
("evt-5", "GiftCardIssued", "order-3002"), # a new type
("evt-6", "PaymentCaptured", "order-3002")]
batch = []
for event_id, event_type, order in EVENTS:
event = EventData(json.dumps({"order": order}))
event.message_id = event_id # unique: the dedupe key
event.correlation_id = order # ties one order together
event.content_type = "application/json"
event.properties = {"event_type": event_type,
"event_version": "1", "source": "shop-api"}
batch.append(event)
send("shop-events", batch)
print("published", len(batch), "events to shop-events")import collections
from ehio import read
HANDLERS = {"OrderPlaced": "reserve stock",
"PaymentCaptured": "post to ledger",
"OrderShipped": "email customer"}
unknown = collections.Counter()
for event in read("shop-events", "router", count=6):
props = {k.decode(): v.decode()
for k, v in event.properties.items()} # bytes
action = HANDLERS.get(props["event_type"]) # body not parsed
if action is None:
unknown[props["event_type"]] += 1
action = "no handler: skipped"
print(event.message_id, event.correlation_id, props["event_type"],
"v" + props["event_version"], "->", action)
print("unknown event types:", dict(unknown))$ python publish_events.py
published 6 events to shop-events
$ python router.py
evt-1 order-3001 OrderPlaced v1 -> reserve stock
evt-2 order-3001 PaymentCaptured v1 -> post to ledger
evt-3 order-3002 OrderPlaced v1 -> reserve stock
evt-4 order-3001 OrderShipped v1 -> email customer
evt-5 order-3002 GiftCardIssued v1 -> no handler: skipped
evt-6 order-3002 PaymentCaptured v1 -> post to ledger
unknown event types: {'GiftCardIssued': 1}Each type goes to its handler: OrderPlaced to reserve stock, PaymentCaptured to post to ledger, OrderShipped to email customer. The gift card event evt-5 has no handler: skipped, and the router counts it: unknown event types: {'GiftCardIssued': 1}. Skipped and counted, not a crash. New types appear all the time, and consumers must tolerate them.
Keep one small envelope for every type, because properties count toward the event size.
Property keys and string values come back as bytes, so decode them in one place.
Question 6: Your manager asks who owns an event contract between teams. How do you manage it?
The producing team owns it, because they own the data. Consumers don't get a veto on every change, but they get a promise: no breaking change without a new event type and a migration plan.
The contract lives in Git next to the producer code, and every pull request runs a compatibility check. Version 3 renames amount to total. Let's check v1 to v2, then v2 to v3, with python check_contract.py v1 v2 and python check_contract.py v2 v3.
{"type": "record", "name": "OrderPlaced", "namespace": "shop",
"fields": [
{"name": "order_id", "type": "long"},
{"name": "customer_id", "type": "string"},
{"name": "total", "type": "double"},
{"name": "currency", "type": "string"},
{"name": "coupon", "type": ["null", "string"], "default": null}
]}import io, random, sys
import fastavro
from fastavro.utils import generate_one
from avro_io import schema
random.seed(7) # sample values don't matter
def can_read(writer, reader):
buf = io.BytesIO()
fastavro.schemaless_writer(buf, writer, generate_one(writer))
try:
fastavro.schemaless_reader(io.BytesIO(buf.getvalue()),
writer, reader)
return "OK"
except fastavro.read.SchemaResolutionError as e:
return f"FAIL ({e})"
old, new = sys.argv[1], sys.argv[2]
backward = can_read(schema(old), schema(new)) # new app, old events
forward = can_read(schema(new), schema(old)) # old app, new events
print(f"{old} -> {new}\n backward: {backward}\n forward: {forward}")
sys.exit(0 if backward == forward == "OK" else 1)$ python check_contract.py v1 v2
v1 -> v2
backward: OK
forward: OK
$ python check_contract.py v2 v3
v2 -> v3
backward: FAIL (No default value for field total in shop.OrderPlaced)
forward: FAIL (No default value for field amount in shop.OrderPlaced)Version 1 to 2 passes both directions: backward: OK and forward: OK. Version 2 to 3 fails both: No default value for field total and No default value for field amount. The build fails, and nothing is registered.
Only after the check passes does the pipeline register the new version:
# CI pipeline, after the compatibility check passed
registry.register_schema("orders", "shop.OrderPlaced",
open("schemas/order_v2.avsc").read(),
"Avro")Access follows the same line:
| Who | Role | Can |
|---|---|---|
| Producer pipeline | Schema Registry Contributor | read, write, delete |
| Consumers | Schema Registry Reader | read and list |
For a real breaking change, we follow three steps:
- Publish a new event type or a new event hub
- Run old and new side by side
- Retire the old one when every consumer has moved
Every schema group needs an owner and a channel consumers can reach. Most contract failures I've seen were a missing conversation, not a missing tool.
Scenario 7: Five teams publish to shared event hubs and changes keep breaking consumers. How do you introduce schemas safely?
In phases, with no big bang. Here is the plan:
- Inventory - one envelope in properties
- Schema groups - per domain, Backward, as code
- Version one - move producers one team at a time
- The gate - check in every pull request
- Measure - rejected and unknown events
First, list what each team sends today, and agree on one envelope: event type, event version, source and a message ID as properties.
Second, create schema groups per domain with Backward compatibility, in Terraform or Bicep, so the setup is reviewed like code. This runs against Azure, not the emulator:
resource "azurerm_eventhub_namespace_schema_group" "orders" {
name = "orders"
namespace_id = azurerm_eventhub_namespace.shop.id
schema_compatibility = "Backward"
schema_type = "Avro"
}Check the tier, because the limits are very different:
| Tier | Schema groups | Versions |
|---|---|---|
| Standard | 1 (plus the default) | 25 |
| Premium | 100 | 1,000 |
| Dedicated | 1,000 | 10,000 |
Third, register today's shapes as version 1, and move producers to the Avro encoder one team at a time. Consumers switch on the content type, so JSON and Avro events can share a hub during the move.
Fourth, turn on the gate. The compatibility check runs in every pull request, only the pipeline can register, and auto_register is off in production.
Finally, measure it. Count rejected and unknown events in every consumer, and alert on them.
A breaking change shows up in a pull request, not in an incident.
Key points to remember
- Schema Registry: schema groups with a compatibility mode stop breaking changes early
- Avro is compact but needs the writer schema; JSON is readable but unenforced
- Backward, Forward, None: add optional fields with defaults
- The schema ID travels in the content type: avro/binary+ID
- Route on event properties; the producer team owns the contract, checked in CI
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 Azurite, run pip install azure-eventhub azure-eventhub-checkpointstoreblob azure-storage-blob confluent-kafka fastavro first, and run it in an empty folder.
Show the full script that writes the files and runs every step
# ========================================================================
# Event Hubs Interview Q&A #7 - Schema Registry and event formats
# 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_07_Code.py.
# Entities in Config.json: orders-json (2 partitions, groups billing); orders (2 partitions, groups billing, analytics); shop-events (2 partitions, groups router)
# 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('ehio.py', 'import logging, os, threading\nfrom azure.eventhub import EventHubConsumerClient, EventHubProducerClient\n\nlogging.getLogger("azure.eventhub").setLevel(logging.ERROR)\nCONN = os.environ["EVENTHUB_CONNECTION_STRING"]\n\n\ndef send(hub, events, partition_id="0"):\n """Send a list of EventData as one batch to one partition."""\n with EventHubProducerClient.from_connection_string(\n CONN, eventhub_name=hub) as producer:\n batch = producer.create_batch(partition_id=partition_id)\n for event in events:\n batch.add(event)\n producer.send_batch(batch)\n\n\ndef read(hub, group, count, timeout=30):\n """Read `count` events from the start, in partition order."""\n client = EventHubConsumerClient.from_connection_string(\n CONN, consumer_group=group, eventhub_name=hub)\n got, lock = [], threading.Lock()\n\n def on_event(pc, event):\n with lock:\n seq = event.sequence_number\n got.append((pc.partition_id, seq, event))\n if len(got) == count:\n threading.Thread(target=client.close).start()\n\n timer = threading.Timer(timeout, client.close)\n timer.start()\n client.receive(on_event, starting_position="-1")\n timer.cancel()\n return [event for _, _, event in sorted(got, key=lambda g: g[:2])]\n\n\ndef body_bytes(event):\n return b"".join(event.body)\n')
write('avro_io.py', """\
import io
import fastavro
def schema(version):
return fastavro.schema.load_schema(f"schemas/order_{version}.avsc")
def encode(record, version):
buf = io.BytesIO()
fastavro.schemaless_writer(buf, schema(version), record)
return buf.getvalue() # Avro binary, no header
def decode(data, writer_version, reader_version):
return fastavro.schemaless_reader(io.BytesIO(data),
schema(writer_version),
schema(reader_version))
""")
write('schemas/order_v1.avsc', """\
{"type": "record", "name": "OrderPlaced", "namespace": "shop",
"fields": [
{"name": "order_id", "type": "long"},
{"name": "customer_id", "type": "string"},
{"name": "amount", "type": "double"},
{"name": "currency", "type": "string"}
]}
""")
write('schemas/order_v2.avsc', """\
{"type": "record", "name": "OrderPlaced", "namespace": "shop",
"fields": [
{"name": "order_id", "type": "long"},
{"name": "customer_id", "type": "string"},
{"name": "amount", "type": "double"},
{"name": "currency", "type": "string"},
{"name": "coupon", "type": ["null", "string"], "default": null}
]}
""")
# Step 1: The producer publishes order 1001 as JSON with an amount field
write('publish_order.py', """\
import json, sys
from azure.eventhub import EventData
from ehio import send
order_id = int(sys.argv[1])
order = {"order_id": order_id, "customer_id": "cust-7",
"amount": 25.50, "currency": "USD"}
event = EventData(json.dumps(order))
event.content_type = "application/json"
send("orders-json", [event])
print("published order", order_id, "fields:", sorted(order))
""")
run('python publish_order.py 1001')
# Step 2: A teammate renames amount to total; the billing consumer rejects the new event
write('publish_order.py', """\
import json, sys
from azure.eventhub import EventData
from ehio import send
order_id = int(sys.argv[1])
order = {"order_id": order_id, "customer_id": "cust-7",
"total": 25.50, "currency": "USD"} # was "amount"
event = EventData(json.dumps(order))
event.content_type = "application/json"
send("orders-json", [event])
print("published order", order_id, "fields:", sorted(order))
""")
write('billing_consumer.py', """\
from ehio import read
REQUIRED = ["order_id", "customer_id", "amount", "currency"]
for event in read("orders-json", "billing", count=2):
order = event.body_as_json()
missing = [f for f in REQUIRED if f not in order]
if missing:
print("order", order["order_id"], "REJECTED: missing", missing)
continue
print("order", order["order_id"],
f"billed {order['amount']:.2f} {order['currency']}")
""")
run('python publish_order.py 1002')
run('python billing_consumer.py')
# Step 3: The same order as JSON and as Avro: what travels in the event body
write('send_formats.py', """\
import json
from azure.eventhub import EventData
from avro_io import encode
from ehio import send
order = {"order_id": 2001, "customer_id": "cust-7",
"amount": 25.50, "currency": "USD"}
as_json = EventData(json.dumps(order))
as_json.content_type = "application/json"
as_avro = EventData(encode(order, "v1")) # values only, no names
as_avro.content_type = "avro/binary"
as_avro.properties = {"schema_version": "v1"}
send("orders", [as_json, as_avro])
print("sent order 2001 twice: as JSON and as Avro")
""")
write('read_formats.py', """\
from ehio import read, body_bytes
for event in read("orders", "analytics", count=2):
body = body_bytes(event)
print(f"{event.content_type:17} {len(body):3} bytes {body[:28]}")
""")
write('size_test.py', """\
import json
from avro_io import encode
orders = [{"order_id": 100000 + i, "customer_id": f"cust-{i % 500}",
"amount": round(10 + i * 0.37, 2), "currency": "USD"}
for i in range(10000)]
json_bytes = sum(len(json.dumps(o).encode()) for o in orders)
avro_bytes = sum(len(encode(o, "v1")) for o in orders)
print("10,000 orders as JSON:", json_bytes, "bytes")
print("10,000 orders as Avro:", avro_bytes, "bytes")
print(f"Avro is {avro_bytes / json_bytes:.0%} of the JSON size")
""")
run('python send_formats.py')
run('python read_formats.py')
run('python size_test.py')
# Step 4: Schema v2 adds an optional coupon field: an old and a new consumer read both versions
write('publish_v2.py', """\
from azure.eventhub import EventData
from avro_io import encode
from ehio import send
order = {"order_id": 2002, "customer_id": "cust-9", "amount": 40.00,
"currency": "USD", "coupon": "FALL10"} # new in v2
event = EventData(encode(order, "v2"))
event.content_type = "avro/binary"
event.properties = {"schema_version": "v2"}
send("orders", [event])
print("producer upgraded: sent order 2002 with schema v2")
""")
write('reader.py', """\
import sys
from avro_io import decode
from ehio import read, body_bytes
reader_version = sys.argv[1] # the schema this app knows
print("consumer with reader schema", reader_version)
for event in read("orders", sys.argv[2], count=3):
if event.content_type != "avro/binary":
continue # the JSON copy
writer = event.properties[b"schema_version"].decode()
order = decode(body_bytes(event), writer, reader_version)
print(" ", writer, order)
""")
run('python publish_v2.py')
run('python reader.py v1 billing')
run('python reader.py v2 analytics')
# Step 5: Decoding Avro with a guessed writer schema
write('wrong_schema.py', """\
from avro_io import decode
from ehio import read, body_bytes
for event in read("orders", "$Default", count=3):
if event.content_type != "avro/binary":
continue
print("event properties:", event.properties)
for guess in ["v1", "v2"]: # pretend we don't know
try:
order = decode(body_bytes(event), guess, guess)
print(f" as {guess}: {order}")
except Exception as e:
print(f" as {guess}: {type(e).__name__}")
""")
run('python wrong_schema.py')
# Step 6: Six events of four types in one event hub; the router reads only the properties
write('publish_events.py', """\
import json
from azure.eventhub import EventData
from ehio import send
EVENTS = [("evt-1", "OrderPlaced", "order-3001"),
("evt-2", "PaymentCaptured", "order-3001"),
("evt-3", "OrderPlaced", "order-3002"),
("evt-4", "OrderShipped", "order-3001"),
("evt-5", "GiftCardIssued", "order-3002"), # a new type
("evt-6", "PaymentCaptured", "order-3002")]
batch = []
for event_id, event_type, order in EVENTS:
event = EventData(json.dumps({"order": order}))
event.message_id = event_id # unique: the dedupe key
event.correlation_id = order # ties one order together
event.content_type = "application/json"
event.properties = {"event_type": event_type,
"event_version": "1", "source": "shop-api"}
batch.append(event)
send("shop-events", batch)
print("published", len(batch), "events to shop-events")
""")
write('router.py', """\
import collections
from ehio import read
HANDLERS = {"OrderPlaced": "reserve stock",
"PaymentCaptured": "post to ledger",
"OrderShipped": "email customer"}
unknown = collections.Counter()
for event in read("shop-events", "router", count=6):
props = {k.decode(): v.decode()
for k, v in event.properties.items()} # bytes
action = HANDLERS.get(props["event_type"]) # body not parsed
if action is None:
unknown[props["event_type"]] += 1
action = "no handler: skipped"
print(event.message_id, event.correlation_id, props["event_type"],
"v" + props["event_version"], "->", action)
print("unknown event types:", dict(unknown))
""")
run('python publish_events.py')
run('python router.py')
# Step 7: A compatibility check in CI: add an optional field, then rename a field
write('schemas/order_v3.avsc', """\
{"type": "record", "name": "OrderPlaced", "namespace": "shop",
"fields": [
{"name": "order_id", "type": "long"},
{"name": "customer_id", "type": "string"},
{"name": "total", "type": "double"},
{"name": "currency", "type": "string"},
{"name": "coupon", "type": ["null", "string"], "default": null}
]}
""")
write('check_contract.py', 'import io, random, sys\nimport fastavro\nfrom fastavro.utils import generate_one\nfrom avro_io import schema\n\nrandom.seed(7) # sample values don\'t matter\n\n\ndef can_read(writer, reader):\n buf = io.BytesIO()\n fastavro.schemaless_writer(buf, writer, generate_one(writer))\n try:\n fastavro.schemaless_reader(io.BytesIO(buf.getvalue()),\n writer, reader)\n return "OK"\n except fastavro.read.SchemaResolutionError as e:\n return f"FAIL ({e})"\n\n\nold, new = sys.argv[1], sys.argv[2]\nbackward = can_read(schema(old), schema(new)) # new app, old events\nforward = can_read(schema(new), schema(old)) # old app, new events\nprint(f"{old} -> {new}\\n backward: {backward}\\n forward: {forward}")\nsys.exit(0 if backward == forward == "OK" else 1)\n')
run('python check_contract.py v1 v2')
run('python check_contract.py v2 v3', expect_error=True) # EXPECTED TO FAIL: the video shows this error
Part 8: Stream Processing with Event Hubs
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.