In this post we are going to cover stream processing with 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: fraud must be flagged within ten seconds of a card payment.
Event Hubs takes in the events. Something else has to read them and do the work: totals per minute, joins with reference data, alerts. That something can be Stream Analytics, Databricks, Azure Functions or Fabric Eventstream. Interviewers ask about this because windows, late events and checkpoints are where streaming jobs quietly give wrong numbers.
Part 8 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: Your manager wants totals every minute from the stream. How do tumbling windows work?
- Question 2: A teammate asks which service should process the stream: Stream Analytics, Databricks, Functions or Fabric. How do you choose?
- Question 3: Events arrive late and out of order. How do event time, watermarks and late arrivals work?
- Question 4: Someone asks how Spark Structured Streaming reads from Event Hubs. What do you set up?
- Question 5: An Azure Function triggered by Event Hubs keeps reprocessing events. How does the trigger checkpoint?
- Question 6: The stream must be joined with reference data. How do you enrich events?
- Scenario 7: Fraud must be flagged within ten seconds of a card payment. What is your streaming design?
- The full script
The short answer
Window by event time, not arrival time: tumbling windows give one result per interval, and a watermark decides when a window closes; keep late events instead of silently dropping them. Pick Stream Analytics, Databricks, Functions or Fabric Eventstream by the shape of the logic, each with its own consumer group.
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 confluent-kafka, so every output is real. Stream Analytics, Spark and Functions don't run on the emulator, so their parts are shown as code. Config.json creates three event hubs: payments with 4 partitions, store-sales with 1 and fraud-alerts with 2. ehio.py holds small send and read helpers, payments.jsonl has twelve card payments with the time they happened in the payload, store_sales.jsonl has seven store sales, and customers.json is our reference data.
{
"UserConfig": {
"NamespaceConfig": [
{
"Type": "EventHub",
"Name": "emulatorNs1",
"Entities": [
{
"Name": "payments",
"PartitionCount": 4,
"ConsumerGroups": [
{
"Name": "totals"
},
{
"Name": "enrich"
},
{
"Name": "fraud"
}
]
},
{
"Name": "store-sales",
"PartitionCount": 1,
"ConsumerGroups": [
{
"Name": "stream"
}
]
},
{
"Name": "fraud-alerts",
"PartitionCount": 2,
"ConsumerGroups": [
{
"Name": "alerting"
}
]
}
]
}
],
"LoggingConfig": {
"Type": "File"
}
}
}import json, logging, os, threading
from datetime import datetime
from azure.eventhub import (EventData, EventHubConsumerClient,
EventHubProducerClient)
logging.getLogger("azure.eventhub").setLevel(logging.ERROR)
CONN = os.environ["EVENTHUB_CONNECTION_STRING"]
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 event_time(payment):
"""The time the payment happened, from the payload."""
return datetime.fromisoformat(payment["event_time"])
def send(hub, items, key):
"""Send dicts as JSON events, partition key = item[key]."""
producer = EventHubProducerClient.from_connection_string(
CONN, eventhub_name=hub)
with producer:
for item in items:
producer.send_event(EventData(json.dumps(item)),
partition_key=item[key]){"payment_id": "pay-01", "card_id": "card-A", "customer_id": "cust-1", "amount": 20.00, "event_time": "2026-10-29T10:00:05Z"}
{"payment_id": "pay-02", "card_id": "card-B", "customer_id": "cust-2", "amount": 35.50, "event_time": "2026-10-29T10:00:18Z"}
{"payment_id": "pay-03", "card_id": "card-C", "customer_id": "cust-3", "amount": 12.25, "event_time": "2026-10-29T10:00:41Z"}
{"payment_id": "pay-04", "card_id": "card-A", "customer_id": "cust-1", "amount": 8.00, "event_time": "2026-10-29T10:00:59Z"}
{"payment_id": "pay-05", "card_id": "card-D", "customer_id": "cust-4", "amount": 60.00, "event_time": "2026-10-29T10:01:02Z"}
{"payment_id": "pay-06", "card_id": "card-B", "customer_id": "cust-2", "amount": 15.00, "event_time": "2026-10-29T10:01:30Z"}
{"payment_id": "pay-07", "card_id": "card-E", "customer_id": "cust-9", "amount": 99.99, "event_time": "2026-10-29T10:01:47Z"}
{"payment_id": "pay-08", "card_id": "card-A", "customer_id": "cust-1", "amount": 5.00, "event_time": "2026-10-29T10:02:10Z"}
{"payment_id": "pay-09", "card_id": "card-F", "customer_id": "cust-5", "amount": 49.00, "event_time": "2026-10-29T10:02:20Z"}
{"payment_id": "pay-10", "card_id": "card-F", "customer_id": "cust-5", "amount": 49.00, "event_time": "2026-10-29T10:02:31Z"}
{"payment_id": "pay-11", "card_id": "card-F", "customer_id": "cust-5", "amount": 49.00, "event_time": "2026-10-29T10:02:39Z"}
{"payment_id": "pay-12", "card_id": "card-C", "customer_id": "cust-3", "amount": 750.00, "event_time": "2026-10-29T10:02:50Z"}{"sale_id": "s-1", "store": "store-12", "amount": 10.00, "event_time": "2026-10-29T10:00:10Z"}
{"sale_id": "s-2", "store": "store-12", "amount": 20.00, "event_time": "2026-10-29T10:00:35Z"}
{"sale_id": "s-3", "store": "store-12", "amount": 5.00, "event_time": "2026-10-29T10:00:25Z"}
{"sale_id": "s-4", "store": "store-12", "amount": 30.00, "event_time": "2026-10-29T10:01:05Z"}
{"sale_id": "s-5", "store": "store-12", "amount": 15.00, "event_time": "2026-10-29T10:01:40Z"}
{"sale_id": "s-6", "store": "store-12", "amount": 7.00, "event_time": "2026-10-29T10:00:50Z"}
{"sale_id": "s-7", "store": "store-12", "amount": 25.00, "event_time": "2026-10-29T10:02:35Z"}{"cust-1": {"name": "Ava", "tier": "gold", "country": "US"},
"cust-2": {"name": "Ben", "tier": "silver", "country": "CA"},
"cust-3": {"name": "Chloe", "tier": "gold", "country": "US"},
"cust-4": {"name": "Dev", "tier": "bronze", "country": "UK"},
"cust-5": {"name": "Eli", "tier": "bronze", "country": "US"}}Question 1: Your manager wants totals every minute from the stream. How do tumbling windows work?
A tumbling window cuts time into fixed, back to back intervals that don't overlap. Every event lands in exactly one window, and each window emits its result when it ends.
Let's try it. We send the twelve card payments into the payments hub, keyed by card_id, with python send_payments.py. Then python tumbling.py puts each payment into the minute of its event time.
import json
from azure.eventhub import EventData, EventHubProducerClient
from ehio import CONN
producer = EventHubProducerClient.from_connection_string(
CONN, eventhub_name="payments")
with producer:
for line in open("payments.jsonl"):
payment = json.loads(line)
producer.send_event(EventData(line.strip()),
partition_key=payment["card_id"])
print("sent 12 card payments, partition key = card_id")import collections
from datetime import timedelta
from ehio import read, event_time
windows = collections.defaultdict(list)
for event in read("payments", "totals", count=12):
payment = event.body_as_json()
t = event_time(payment) # event time, not arrival
start = t.replace(second=0, microsecond=0) # one-minute window
windows[start].append(payment["amount"])
for start in sorted(windows):
end, amounts = start + timedelta(minutes=1), windows[start]
print(f"window {start:%H:%M} to {end:%H:%M}:",
f"{len(amounts)} payments, total {sum(amounts):7.2f}")$ python send_payments.py
sent 12 card payments, partition key = card_id
$ python tumbling.py
window 10:00 to 10:01: 4 payments, total 75.75
window 10:01 to 10:02: 3 payments, total 174.99
window 10:02 to 10:03: 5 payments, total 902.00We get three windows. window 10:00 to 10:01: 4 payments, total 75.75, then 3 payments, total 174.99, then 5 payments, total 902.00. Each payment is counted once.
In Stream Analytics it is one query: group by a tumbling window of one minute.
SELECT System.Timestamp() AS window_end,
COUNT(*) AS payments,
SUM(amount) AS total
INTO [minute-totals]
FROM [payments] TIMESTAMP BY event_time
GROUP BY TumblingWindow(minute, 1)Without TIMESTAMP BY, the job uses the arrival time, EventEnqueuedUtcTime. TIMESTAMP BY event_time uses the time in the payload instead.
Always ask which clock the business means. A payment made at 10:59 that arrives at 11:01 belongs to the earlier minute, and only event time gets that right.
Question 2: A teammate asks which service should process the stream: Stream Analytics, Databricks, Functions or Fabric. How do you choose?
By the shape of the logic and the team that runs it. All four read from Event Hubs, and each job should use its own consumer group.
| Service | Best for | Watch out |
|---|---|---|
| Stream Analytics | SQL windows, joins, reference data; no-code editor | logic must fit the query language |
| Databricks (Spark) | complex logic, ML, Delta tables; Kafka endpoint | you run and tune the clusters |
| Azure Functions | actions per event or batch: APIs, notifications | no windows built in; at least once |
| Fabric Eventstream | no-code: Eventhouse, Lakehouse, Activator | best when the team lives in Fabric |
Let's go through them one by one. Stream Analytics is for SQL shaped work: windows, joins and reference data, fully managed, with a no-code editor. Databricks is for complex logic, machine learning and Delta tables. It reads through the Kafka endpoint, and we run the clusters.
Azure Functions is for actions per event or batch, like calling an API or sending a notification. It has no windows built in. Fabric Eventstream is no-code: it brings events into Fabric and routes them to an Eventhouse, a Lakehouse or Activator.
One stream often feeds several at once. That is what consumer groups are for:
payments- one event hub- Stream Analytics - consumer group
asa-totals, for live totals - Databricks - consumer group
lakehouse - Function - consumer group
notify, for notifications
Picking the tool the team knows and forcing the problem into it. A minute total doesn't need a Spark cluster.
Question 3: Events arrive late and out of order. How do event time, watermarks and late arrivals work?
Event time is when it happened. Arrival time is when Event Hubs got it. A watermark marks how far event time has progressed: commonly the latest event time seen, minus an allowed delay. When it passes a window's end, the window closes.
Let's see it. Seven sales from store-12 arrive out of order in one partition, and our consumer allows 30 seconds of delay. We run python send_sales.py and python watermark.py.
from azure.eventhub import EventData, EventHubProducerClient
from ehio import CONN
producer = EventHubProducerClient.from_connection_string(
CONN, eventhub_name="store-sales")
with producer:
for line in open("store_sales.jsonl"): # in ARRIVAL order
producer.send_event(EventData(line.strip()), partition_id="0")
print("sent 7 sales from store-12, in the order they arrived")import collections
from datetime import timedelta
from ehio import read, event_time
DELAY = timedelta(seconds=30) # how late an event may arrive
totals, closed = collections.defaultdict(float), set()
watermark = None
for event in read("store-sales", "stream", count=7): # arrival order
sale = event.body_as_json()
t = event_time(sale)
window = t.replace(second=0)
if window in closed:
print(f"{sale['sale_id']} {t:%H:%M:%S} LATE: window",
f"{window:%H:%M} already closed -> late_sales")
continue
totals[window] += sale["amount"]
candidate = t - DELAY # max event time seen - delay
watermark = max(watermark or candidate, candidate)
print(f"{sale['sale_id']} {t:%H:%M:%S} -> window {window:%H:%M},",
f"watermark {watermark:%H:%M:%S}")
for w in sorted(totals):
if w not in closed and w + timedelta(minutes=1) <= watermark:
closed.add(w)
print(f" CLOSE window {w:%H:%M}: total {totals[w]:.2f}")
still_open = [f"{w:%H:%M}" for w in totals if w not in closed]
print("still open, waiting for the watermark:", still_open)$ python send_sales.py
sent 7 sales from store-12, in the order they arrived
$ python watermark.py
s-1 10:00:10 -> window 10:00, watermark 09:59:40
s-2 10:00:35 -> window 10:00, watermark 10:00:05
s-3 10:00:25 -> window 10:00, watermark 10:00:05
s-4 10:01:05 -> window 10:01, watermark 10:00:35
s-5 10:01:40 -> window 10:01, watermark 10:01:10
CLOSE window 10:00: total 35.00
s-6 10:00:50 LATE: window 10:00 already closed -> late_sales
s-7 10:02:35 -> window 10:02, watermark 10:02:05
CLOSE window 10:01: total 45.00
still open, waiting for the watermark: ['10:02']Let's walk through it:
s-3at10:00:25arrives afters-2but is older. It is inside the delay, so it is accepted intowindow 10:00.s-5moves the watermark to10:01:10, past 10:01, soCLOSE window 10:00: total 35.00.- Then
s-6arrives, from10:00:50. Its window is closed, so it goes tolate_salesinstead of disappearing. s-7closeswindow 10:01: total 45.00, and10:02isstill open, waiting for the watermark.
Stream Analytics has the same idea as job settings: late arrival tolerance, 5 seconds by default, and out-of-order tolerance, 0 by default. Events outside them are dropped or adjusted. In Spark, it is withWatermark on the event time column.
The trade-off is completeness against latency. A longer delay is more complete, but later.
Measure how late events really arrive, and keep the raw events or a late output, never a silent drop.
Question 4: Someone asks how Spark Structured Streaming reads from Event Hubs. What do you set up?
Spark reads Event Hubs through its Kafka endpoint, with the built-in kafka source. On Databricks, the bootstrap server is your namespace on port 9093, with SASL_SSL. The event hub is the Kafka topic.
Before Spark, I check the endpoint with a plain Kafka client. On the emulator it is port 9092 without TLS. The user name is $ConnectionString, the password is the connection string, and the topic is the event hub. We run python kafka_read.py.
import json, os
from confluent_kafka import Consumer
consumer = Consumer({
"bootstrap.servers": "localhost:9092", # Azure: NAMESPACE:9093
"security.protocol": "SASL_PLAINTEXT", # Azure: SASL_SSL
"sasl.mechanism": "PLAIN",
"sasl.username": "$ConnectionString",
"sasl.password": os.environ["EVENTHUB_CONNECTION_STRING"],
"group.id": "spark-check", "auto.offset.reset": "earliest",
"enable.auto.commit": False}) # like Spark: no group commits
consumer.subscribe(["payments"]) # topic = event hub
rows = []
try:
for _ in range(30): # at most 30 polls
msg = consumer.poll(1)
if msg is not None and not msg.error():
pay_id = json.loads(msg.value())["payment_id"]
rows.append((msg.partition(), msg.offset(), pay_id))
if len(rows) == 12:
break
finally:
consumer.close()
for part in sorted({r[0] for r in rows}):
got = [(off, pay) for p, off, pay in sorted(rows) if p == part]
print(f"partition {part}, offsets {got[0][0]}-{got[-1][0]}:",
" ".join(pay for _, pay in got))$ python kafka_read.py
partition 0, offsets 0-1: pay-03 pay-12
partition 1, offsets 0-6: pay-01 pay-04 pay-07 pay-08 pay-09 pay-10 pay-11
partition 2, offsets 0-0: pay-05
partition 3, offsets 0-1: pay-02 pay-06All twelve payments come back, across four partitions: for example partition 1, offsets 0-6 holds seven of them, from pay-01 to pay-11.
The Spark job uses the same settings, with the kafkashaded prefix on the login module. It parses the JSON value, sets a watermark on event time, and writes one minute totals to a Delta table. This runs on Databricks, not on the emulator:
from pyspark.sql import functions as F
conn = dbutils.secrets.get("eventhubs", "payments-listen")
jaas = ("kafkashaded.org.apache.kafka.common.security.plain."
'PlainLoginModule required username="$ConnectionString" '
f'password="{conn}";')
raw = (spark.readStream.format("kafka")
.option("kafka.bootstrap.servers",
"tbshop-ns.servicebus.windows.net:9093")
.option("subscribe", "payments")
.option("kafka.security.protocol", "SASL_SSL")
.option("kafka.sasl.mechanism", "PLAIN")
.option("kafka.sasl.jaas.config", jaas)
.option("startingOffsets", "earliest")
.load())
schema = "payment_id STRING, amount DOUBLE, event_time TIMESTAMP"
pays = (raw.select(F.from_json(F.col("value").cast("string"), schema)
.alias("p")).select("p.*"))
totals = (pays.withWatermark("event_time", "30 seconds")
.groupBy(F.window("event_time", "1 minute"))
.agg(F.count("*").alias("payments"),
F.sum("amount").alias("total")))
(totals.writeStream.outputMode("append")
.option("checkpointLocation", "/Volumes/shop/stream/chk/totals")
.toTable("shop.stream.minute_totals"))Offsets live in the checkpoint location, not in consumer group commits, so a restart resumes from the checkpoint. Never share a checkpoint location between two queries.
Use Microsoft Entra ID instead of a connection string: a service principal with OAUTHBEARER, or Unity Catalog service credentials on Databricks Runtime 16.1 and later.
Question 5: An Azure Function triggered by Event Hubs keeps reprocessing events. How does the trigger checkpoint?
The trigger keeps a pointer per partition in the storage account from the AzureWebJobsStorage setting. When an execution completes, with or without exceptions, it advances the pointer and saves a checkpoint.
So steady reprocessing usually means executions aren't completing. Here are the usual causes:
- Timeout - the execution never completes: same events again
- Instance crash - the pointer never moves
- Retry policy - holds the pointer until all retries finish
The fix: keep each batch well inside the timeout, make the writes idempotent, keyed by payment_id, and catch errors per event, parking failures in a queue so the stream keeps moving. The Functions trigger is not part of our emulator setup, so here it is as code:
import json
import azure.functions as func
app = func.FunctionApp()
@app.event_hub_message_trigger(
arg_name="events", event_hub_name="payments",
connection="EventHubConnection", consumer_group="scoring",
cardinality="many")
def score_payments(events: list[func.EventHubEvent]):
for event in events:
payment = json.loads(event.get_body())
try:
upsert_score(payment) # idempotent: payment_id key
except Exception:
park(payment) # to a queue, keep moving
# returns: the trigger checkpoints this batchCheck host.json too. batchCheckpointFrequency is 1 by default, and raising it replays more after a restart.
{
"version": "2.0",
"extensions": {
"eventHubs": {
"maxEventBatchSize": 100,
"batchCheckpointFrequency": 1,
"prefetchCount": 300,
"initialOffsetOptions": {
"type": "fromStart"
}
}
}
}If we lose the checkpoints, say with a new storage account, the trigger starts from the beginning. And give the function its own consumer group.
Exceptions still advance the pointer, so a function that throws without handling the error loses those events. Catch, then park.
Question 6: The stream must be joined with reference data. How do you enrich events?
Keep the reference data next to the processor, in memory, and join each event against it. Don't call a database for every event: that is where the latency and the throttling come from.
Our enrichment consumer loads customers.json once, then looks up each payment's customer: name, tier and country. We run python enrich.py.
import json
from ehio import read
CUSTOMERS = json.load(open("customers.json")) # loaded once
UNKNOWN = {"name": "?", "tier": "unknown", "country": "?"}
rows = []
for event in read("payments", "enrich", count=12):
pay = event.body_as_json()
cust = CUSTOMERS.get(pay["customer_id"], UNKNOWN) # a left join
rows.append((pay["payment_id"], pay["customer_id"],
f"{pay['amount']:7.2f}", "->", cust["name"],
cust["tier"], cust["country"]))
for row in sorted(rows):
print(*row)$ python enrich.py
pay-01 cust-1 20.00 -> Ava gold US
pay-02 cust-2 35.50 -> Ben silver CA
pay-03 cust-3 12.25 -> Chloe gold US
pay-04 cust-1 8.00 -> Ava gold US
pay-05 cust-4 60.00 -> Dev bronze UK
pay-06 cust-2 15.00 -> Ben silver CA
pay-07 cust-9 99.99 -> ? unknown ?
pay-08 cust-1 5.00 -> Ava gold US
pay-09 cust-5 49.00 -> Eli bronze US
pay-10 cust-5 49.00 -> Eli bronze US
pay-11 cust-5 49.00 -> Eli bronze US
pay-12 cust-3 750.00 -> Chloe gold USEvery payment gets its customer, like pay-01 cust-1 20.00 -> Ava gold US. It is a left join, so pay-07, from unknown cust-9, still flows through, marked ? unknown ?, instead of being dropped.
In Stream Analytics, reference data comes from Blob Storage, ADLS Gen2 or Azure SQL Database, joined in the query:
SELECT p.payment_id, p.amount, c.tier, c.country
INTO [enriched-payments]
FROM [payments] p TIMESTAMP BY event_time
LEFT OUTER JOIN [customers] c
ON p.customer_id = c.customer_idBlob reference data is checked for refreshes every minute, and works best under 300 MB. In Spark, it is a stream to static join against a Delta table, which picks up new table versions as micro-batches run. Either way, refresh reference data on a schedule, not per event.
Count unknown lookups and alert on them. A spike usually means stale reference data, not vanished customers.
Scenario 7: Fraud must be flagged within ten seconds of a card payment. What is your streaming design?
Payments go into an event hub keyed by card_id, so each card's payments stay in order. A stream processor with its own consumer group scores each payment by event time and publishes alerts to a second event hub. Here is the flow:
payments- partition keycard_id- Fraud rules - consumer group
fraud, by event time fraud-alerts- an event hub- Alerting - block, notify
Let's run a small, real version of the rules. It keeps each card's payments from the last 60 seconds. Three payments in that window, or one payment over 500, raise an alert. We run python fraud.py, then python read_alerts.py.
from collections import defaultdict
from datetime import timedelta
from ehio import read, send, event_time
WINDOW, MAX_PAYMENTS, BIG = timedelta(seconds=60), 3, 500.00
recent = defaultdict(list) # card_id -> recent event times
alerts = []
for event in read("payments", "fraud", count=12): # in order per card
pay = event.body_as_json()
t, card = event_time(pay), pay["card_id"]
recent[card] = [x for x in recent[card] if t - x < WINDOW] + [t]
reason = None
if len(recent[card]) >= MAX_PAYMENTS:
reason = f"{len(recent[card])} payments within 60 s"
elif pay["amount"] > BIG:
reason = f"amount {pay['amount']:.2f} over {BIG:.0f}"
if reason:
alerts.append({"payment_id": pay["payment_id"],
"card_id": card, "reason": reason})
send("fraud-alerts", alerts, key="card_id")
print("checked 12 payments, sent", len(alerts), "alerts")from ehio import read
events = read("fraud-alerts", "alerting", count=2)
alerts = [event.body_as_json() for event in events]
for a in sorted(alerts, key=lambda a: a["payment_id"]):
print("ALERT", a["payment_id"], a["card_id"], "-", a["reason"])$ python fraud.py
checked 12 payments, sent 2 alerts
$ python read_alerts.py
ALERT pay-11 card-F - 3 payments within 60 s
ALERT pay-12 card-C - amount 750.00 over 500It prints checked 12 payments, sent 2 alerts. The alerting consumer reads both from the fraud-alerts hub: ALERT pay-11 card-F - 3 payments within 60 s and ALERT pay-12 card-C - amount 750.00 over 500.
For a ten second budget, keep every hop short:
- Always running processor, no batch schedule
- Watermark delay of seconds, not minutes
- Alerts as events that trigger action, not rows someone polls
Plan for restarts too. The engine must checkpoint per card state with its position, and it may score a payment twice.
Alerts carry the payment_id, so the alert consumer dedupes on it.
Key points to remember
- Tumbling windows by event time: one result per interval, each event once
- Choose Stream Analytics, Databricks, Functions or Fabric by the logic; one consumer group each
- A watermark closes windows; keep late events, never drop them silently
- Spark reads through the Kafka endpoint; offsets live in the checkpoint
- Functions checkpoint after every completed execution; reference data joins in memory
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 #8 - Stream processing with Event Hubs
# 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_08_Code.py.
# Entities in Config.json: payments (4 partitions, groups totals, enrich, fraud); store-sales (1 partitions, groups stream); fraud-alerts (2 partitions, groups alerting)
# 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 json, logging, os, threading\nfrom datetime import datetime\nfrom azure.eventhub import (EventData, EventHubConsumerClient,\n EventHubProducerClient)\n\nlogging.getLogger("azure.eventhub").setLevel(logging.ERROR)\nCONN = os.environ["EVENTHUB_CONNECTION_STRING"]\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 event_time(payment):\n """The time the payment happened, from the payload."""\n return datetime.fromisoformat(payment["event_time"])\n\n\ndef send(hub, items, key):\n """Send dicts as JSON events, partition key = item[key]."""\n producer = EventHubProducerClient.from_connection_string(\n CONN, eventhub_name=hub)\n with producer:\n for item in items:\n producer.send_event(EventData(json.dumps(item)),\n partition_key=item[key])\n')
write('payments.jsonl', """\
{"payment_id": "pay-01", "card_id": "card-A", "customer_id": "cust-1", "amount": 20.00, "event_time": "2026-10-29T10:00:05Z"}
{"payment_id": "pay-02", "card_id": "card-B", "customer_id": "cust-2", "amount": 35.50, "event_time": "2026-10-29T10:00:18Z"}
{"payment_id": "pay-03", "card_id": "card-C", "customer_id": "cust-3", "amount": 12.25, "event_time": "2026-10-29T10:00:41Z"}
{"payment_id": "pay-04", "card_id": "card-A", "customer_id": "cust-1", "amount": 8.00, "event_time": "2026-10-29T10:00:59Z"}
{"payment_id": "pay-05", "card_id": "card-D", "customer_id": "cust-4", "amount": 60.00, "event_time": "2026-10-29T10:01:02Z"}
{"payment_id": "pay-06", "card_id": "card-B", "customer_id": "cust-2", "amount": 15.00, "event_time": "2026-10-29T10:01:30Z"}
{"payment_id": "pay-07", "card_id": "card-E", "customer_id": "cust-9", "amount": 99.99, "event_time": "2026-10-29T10:01:47Z"}
{"payment_id": "pay-08", "card_id": "card-A", "customer_id": "cust-1", "amount": 5.00, "event_time": "2026-10-29T10:02:10Z"}
{"payment_id": "pay-09", "card_id": "card-F", "customer_id": "cust-5", "amount": 49.00, "event_time": "2026-10-29T10:02:20Z"}
{"payment_id": "pay-10", "card_id": "card-F", "customer_id": "cust-5", "amount": 49.00, "event_time": "2026-10-29T10:02:31Z"}
{"payment_id": "pay-11", "card_id": "card-F", "customer_id": "cust-5", "amount": 49.00, "event_time": "2026-10-29T10:02:39Z"}
{"payment_id": "pay-12", "card_id": "card-C", "customer_id": "cust-3", "amount": 750.00, "event_time": "2026-10-29T10:02:50Z"}
""")
write('store_sales.jsonl', """\
{"sale_id": "s-1", "store": "store-12", "amount": 10.00, "event_time": "2026-10-29T10:00:10Z"}
{"sale_id": "s-2", "store": "store-12", "amount": 20.00, "event_time": "2026-10-29T10:00:35Z"}
{"sale_id": "s-3", "store": "store-12", "amount": 5.00, "event_time": "2026-10-29T10:00:25Z"}
{"sale_id": "s-4", "store": "store-12", "amount": 30.00, "event_time": "2026-10-29T10:01:05Z"}
{"sale_id": "s-5", "store": "store-12", "amount": 15.00, "event_time": "2026-10-29T10:01:40Z"}
{"sale_id": "s-6", "store": "store-12", "amount": 7.00, "event_time": "2026-10-29T10:00:50Z"}
{"sale_id": "s-7", "store": "store-12", "amount": 25.00, "event_time": "2026-10-29T10:02:35Z"}
""")
write('customers.json', """\
{"cust-1": {"name": "Ava", "tier": "gold", "country": "US"},
"cust-2": {"name": "Ben", "tier": "silver", "country": "CA"},
"cust-3": {"name": "Chloe", "tier": "gold", "country": "US"},
"cust-4": {"name": "Dev", "tier": "bronze", "country": "UK"},
"cust-5": {"name": "Eli", "tier": "bronze", "country": "US"}}
""")
# Step 1: Twelve card payments with event times in the payload; one-minute tumbling totals by event time
write('send_payments.py', """\
import json
from azure.eventhub import EventData, EventHubProducerClient
from ehio import CONN
producer = EventHubProducerClient.from_connection_string(
CONN, eventhub_name="payments")
with producer:
for line in open("payments.jsonl"):
payment = json.loads(line)
producer.send_event(EventData(line.strip()),
partition_key=payment["card_id"])
print("sent 12 card payments, partition key = card_id")
""")
write('tumbling.py', """\
import collections
from datetime import timedelta
from ehio import read, event_time
windows = collections.defaultdict(list)
for event in read("payments", "totals", count=12):
payment = event.body_as_json()
t = event_time(payment) # event time, not arrival
start = t.replace(second=0, microsecond=0) # one-minute window
windows[start].append(payment["amount"])
for start in sorted(windows):
end, amounts = start + timedelta(minutes=1), windows[start]
print(f"window {start:%H:%M} to {end:%H:%M}:",
f"{len(amounts)} payments, total {sum(amounts):7.2f}")
""")
run('python send_payments.py')
run('python tumbling.py')
# Step 2: Seven store sales arrive out of order; a 30 second watermark closes windows and catches a late sale
write('send_sales.py', """\
from azure.eventhub import EventData, EventHubProducerClient
from ehio import CONN
producer = EventHubProducerClient.from_connection_string(
CONN, eventhub_name="store-sales")
with producer:
for line in open("store_sales.jsonl"): # in ARRIVAL order
producer.send_event(EventData(line.strip()), partition_id="0")
print("sent 7 sales from store-12, in the order they arrived")
""")
write('watermark.py', """\
import collections
from datetime import timedelta
from ehio import read, event_time
DELAY = timedelta(seconds=30) # how late an event may arrive
totals, closed = collections.defaultdict(float), set()
watermark = None
for event in read("store-sales", "stream", count=7): # arrival order
sale = event.body_as_json()
t = event_time(sale)
window = t.replace(second=0)
if window in closed:
print(f"{sale['sale_id']} {t:%H:%M:%S} LATE: window",
f"{window:%H:%M} already closed -> late_sales")
continue
totals[window] += sale["amount"]
candidate = t - DELAY # max event time seen - delay
watermark = max(watermark or candidate, candidate)
print(f"{sale['sale_id']} {t:%H:%M:%S} -> window {window:%H:%M},",
f"watermark {watermark:%H:%M:%S}")
for w in sorted(totals):
if w not in closed and w + timedelta(minutes=1) <= watermark:
closed.add(w)
print(f" CLOSE window {w:%H:%M}: total {totals[w]:.2f}")
still_open = [f"{w:%H:%M}" for w in totals if w not in closed]
print("still open, waiting for the watermark:", still_open)
""")
run('python send_sales.py')
run('python watermark.py')
# Step 3: The Kafka endpoint Spark uses: read the payments event hub with a Kafka client
write('kafka_read.py', """\
import json, os
from confluent_kafka import Consumer
consumer = Consumer({
"bootstrap.servers": "localhost:9092", # Azure: NAMESPACE:9093
"security.protocol": "SASL_PLAINTEXT", # Azure: SASL_SSL
"sasl.mechanism": "PLAIN",
"sasl.username": "$ConnectionString",
"sasl.password": os.environ["EVENTHUB_CONNECTION_STRING"],
"group.id": "spark-check", "auto.offset.reset": "earliest",
"enable.auto.commit": False}) # like Spark: no group commits
consumer.subscribe(["payments"]) # topic = event hub
rows = []
try:
for _ in range(30): # at most 30 polls
msg = consumer.poll(1)
if msg is not None and not msg.error():
pay_id = json.loads(msg.value())["payment_id"]
rows.append((msg.partition(), msg.offset(), pay_id))
if len(rows) == 12:
break
finally:
consumer.close()
for part in sorted({r[0] for r in rows}):
got = [(off, pay) for p, off, pay in sorted(rows) if p == part]
print(f"partition {part}, offsets {got[0][0]}-{got[-1][0]}:",
" ".join(pay for _, pay in got))
""")
run('python kafka_read.py')
# Step 4: Enrich payments with customer reference data held in memory (a left join)
write('enrich.py', """\
import json
from ehio import read
CUSTOMERS = json.load(open("customers.json")) # loaded once
UNKNOWN = {"name": "?", "tier": "unknown", "country": "?"}
rows = []
for event in read("payments", "enrich", count=12):
pay = event.body_as_json()
cust = CUSTOMERS.get(pay["customer_id"], UNKNOWN) # a left join
rows.append((pay["payment_id"], pay["customer_id"],
f"{pay['amount']:7.2f}", "->", cust["name"],
cust["tier"], cust["country"]))
for row in sorted(rows):
print(*row)
""")
run('python enrich.py')
# Step 5: Fraud rules per card by event time; alerts go to a second event hub
write('fraud.py', """\
from collections import defaultdict
from datetime import timedelta
from ehio import read, send, event_time
WINDOW, MAX_PAYMENTS, BIG = timedelta(seconds=60), 3, 500.00
recent = defaultdict(list) # card_id -> recent event times
alerts = []
for event in read("payments", "fraud", count=12): # in order per card
pay = event.body_as_json()
t, card = event_time(pay), pay["card_id"]
recent[card] = [x for x in recent[card] if t - x < WINDOW] + [t]
reason = None
if len(recent[card]) >= MAX_PAYMENTS:
reason = f"{len(recent[card])} payments within 60 s"
elif pay["amount"] > BIG:
reason = f"amount {pay['amount']:.2f} over {BIG:.0f}"
if reason:
alerts.append({"payment_id": pay["payment_id"],
"card_id": card, "reason": reason})
send("fraud-alerts", alerts, key="card_id")
print("checked 12 payments, sent", len(alerts), "alerts")
""")
write('read_alerts.py', """\
from ehio import read
events = read("fraud-alerts", "alerting", count=2)
alerts = [event.body_as_json() for event in events]
for a in sorted(alerts, key=lambda a: a["payment_id"]):
print("ALERT", a["payment_id"], a["card_id"], "-", a["reason"])
""")
run('python fraud.py')
run('python read_alerts.py')
Part 9: Security, Networking and Disaster Recovery
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.