In this post we are going to cover the change feed and event-driven pipelines in Azure Cosmos DB. These are 7 real Cosmos DB interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario: every order change must flow to the data lake and a search index within a minute.
The change feed is a record of the changes to a container, in the order they happen, that your code can read. It's how other systems react to new and updated data without querying the container again. Interviewers ask about it because almost every data engineering job on Cosmos DB touches it, and the tricky parts are the details: where a reader starts, how it survives a restart, and what happens to deletes.
Part 7 of the Real Time Azure Cosmos DB Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the project
- Question 1: A teammate asks what the change feed is. How do you explain it?
- Question 2: Someone asks how to read the change feed from the start or from now. How do you control the starting point?
- Question 3: A pipeline must not miss changes after a restart. How do continuation tokens and leases work?
- Question 4: A teammate asks whether deletes appear in the change feed. What do you tell them?
- Question 5: Several workers must process the change feed in parallel. How does the change feed processor scale?
- Question 6: Someone wants an Azure Function to react to every new order. How does the Cosmos DB trigger work?
- Scenario 7: Every order change must flow to the data lake and a search index within a minute. How do you build it?
- The full script
The short answer
The change feed is a persistent, ordered record of creates and updates in a container. The default latest version mode keeps only the newest version and no deletes; all versions and deletes mode adds them (continuous backup). Readers keep their position in a continuation token (pull model) or leases (change feed processor, Azure Functions) and checkpoint after the work is done.
Let's set up the project
We run the official Azure Cosmos DB Linux emulator locally, and small Python scripts that use the azure-cosmos client, so every output is real. Features the emulator doesn't support are shown as code. common.py creates the client from the COSMOS_ENDPOINT and COSMOS_KEY environment variables, creates the shop database, and has three helpers: container makes a container partitioned by /customerId, order builds an order with an id, a customer and a status, and biz drops the system properties that start with an underscore before we print.
import os
from azure.cosmos import CosmosClient, PartitionKey
client = CosmosClient(os.environ["COSMOS_ENDPOINT"],
credential=os.environ["COSMOS_KEY"])
db = client.create_database_if_not_exists("shop")
def container(name, **options):
return db.create_container_if_not_exists(
name, partition_key=PartitionKey(path="/customerId"),
**options)
def order(oid, customer, status):
return {"id": oid, "customerId": customer, "status": status}
def biz(item): # drop _rid, _etag, _ts, _lsn ...
return {k: v for k, v in item.items() if not k.startswith("_")}Question 1: A teammate asks what the change feed is. How do you explain it?
The change feed is a persistent record of the changes to a container, in the order they happen. It's on by default. Your code reads each created or updated item, so other systems can react without querying the container again.
The shop app writes orders to the orders container, partitioned by /customerId. Readers such as Functions, the change feed processor or the pull model then read the feed.
Let's try it. writes.py makes six writes to the orders container: three new orders, order o1 paid and then shipped, and order o2 deleted. Then it reads the change feed from the beginning. We run python writes.py.
from common import container, order, biz
orders = container("orders")
orders.create_item(order("o1", "c1", "new"))
orders.create_item(order("o2", "c2", "new"))
orders.create_item(order("o3", "c1", "new"))
orders.upsert_item(order("o1", "c1", "paid"))
orders.upsert_item(order("o1", "c1", "shipped"))
orders.delete_item("o2", partition_key="c2")
print("6 writes: 3 creates, o1 updated twice, o2 deleted")
feed = list(orders.query_items_change_feed(start_time="Beginning"))
print("change feed from the beginning:", len(feed), "items")
for d in feed:
print(" ", biz(d))
print("o2 in the feed:", any(d["id"] == "o2" for d in feed))$ python writes.py
6 writes: 3 creates, o1 updated twice, o2 deleted
change feed from the beginning: 2 items
{'id': 'o3', 'customerId': 'c1', 'status': 'new'}
{'id': 'o1', 'customerId': 'c1', 'status': 'shipped'}
o2 in the feed: FalseWe get 2 items, not six. Order o3 as new, and order o1 only as shipped. And o2 in the feed: False. Three things to notice:
- Latest version mode - only the newest version
- No deletes - o2 is simply missing
- Ordered per partition key - not across keys
Order o1 now sits after o3, at the position of its latest change.
The change feed is the backbone of event-driven work: syncing copies, feeding the lake, triggering functions. The only cost is the RUs of the reads, plus a lease container if you use one.
Question 2: Someone asks how to read the change feed from the start or from now. How do you control the starting point?
You choose the start once, when a reader begins: from the beginning, from a point in time, from now, or from a saved continuation token. In Python it's the start_time argument of query_items_change_feed, and now is the default.
| Start from | Python argument | What you get |
|---|---|---|
| the beginning | start_time="Beginning" | the latest version of every item |
| a point in time | start_time=datetime | changes after it (about 5 s precision on Azure) |
| now (the default) | start_time="Now" | only changes from here on |
| a saved token | continuation=tok | exactly where the reader stopped |
Let's try three of them. startpoints.py marks a time, waits, and adds order o4. Then it reads from the beginning and from the point in time. For from now, the first read only fixes the position: the code keeps its token and ignores its rows. Then order o5 arrives, and we read with that token. We run python startpoints.py.
import time
from datetime import datetime, timezone
from common import container, order
orders = container("orders")
def ids(feed):
return [d["id"] for d in feed]
time.sleep(2)
mark = datetime.now(timezone.utc) # a point in time
time.sleep(2)
orders.create_item(order("o4", "c3", "new"))
print("from the beginning:",
ids(orders.query_items_change_feed(start_time="Beginning")))
print("from the point in time:",
ids(orders.query_items_change_feed(start_time=mark)))
now = orders.query_items_change_feed(start_time="Now")
list(now) # this read only fixes the position
tok = orders.client_connection.last_response_headers["etag"]
orders.create_item(order("o5", "c2", "new"))
print("from now, with its token:",
ids(orders.query_items_change_feed(continuation=tok)))$ python startpoints.py
from the beginning: ['o3', 'o1', 'o4']
from the point in time: ['o4']
from now, with its token: ['o5']From the beginning returns ['o3', 'o1', 'o4']: the latest version of everything still there. From the point in time returns only ['o4']. And the read with the token from now returns exactly ['o5'].
On Azure, start time precision is about five seconds, so a point in time is approximate. It's good for replays.
A reader that starts from now on every deploy. Everything written while it was down is skipped. Start from now once, then always resume from the saved token.
Question 3: A pipeline must not miss changes after a restart. How do continuation tokens and leases work?
The reader stores its position. In the pull model, that's a continuation token you save yourself. With the change feed processor or Azure Functions, it's a lease document in a lease container, updated as batches finish.
| Reader | Where the position lives | Who saves it |
|---|---|---|
| pull model (Python) | a continuation token | your code: file, blob, table |
| change feed processor | lease documents in a lease container | the processor, per batch |
| Azure Functions trigger | the lease container (default: leases) | the Functions host |
Let's build a consumer. consumer.py has no token yet, so it starts from the beginning and saves the token to a file. Then app.py plays the shop and writes order o6 and marks o1 delivered. We run python consumer.py and python app.py.
import os
from common import container
orders = container("orders")
TOKEN = "orders.token" # the reader's position
tok = open(TOKEN).read() if os.path.exists(TOKEN) else None
print("start from:", "the saved token" if tok else "the beginning")
kw = {"continuation": tok} if tok else {"start_time": "Beginning"}
pager = orders.query_items_change_feed(**kw).by_page()
n = 0
for page in pager:
for d in page:
print(" ", d["id"], d["status"]) # process the change
n += 1
if pager.continuation_token: # None: nothing new
open(TOKEN, "w").write(pager.continuation_token)
print(f"changes: {n}, token saved")
else:
print(f"changes: {n}, nothing new")from common import container, order
orders = container("orders")
orders.create_item(order("o6", "c3", "new"))
orders.upsert_item(order("o1", "c1", "delivered"))
print("shop wrote: o6 new, o1 delivered")$ python consumer.py
start from: the beginning
o3 new
o1 shipped
o4 new
o5 new
changes: 4, token saved
$ python app.py
shop wrote: o6 new, o1 deliveredThe first run starts from the beginning and gets four changes, o3, o1, o4 and o5, then prints changes: 4, token saved. The shop then writes o6 new, o1 delivered.
Now we restart the emulator, like a crashed host. The data is kept. Then we run python consumer.py twice.
$ python consumer.py
start from: the saved token
o6 new
o1 delivered
changes: 2, token saved
$ python consumer.py
start from: the saved token
changes: 0, nothing newThe consumer reads its token file and gets exactly the two new changes, o6 new and o1 delivered. The next run prints changes: 0, nothing new. Nothing missed, nothing repeated.
Two facts from the docs. Latest version tokens never expire while the container exists. And each change appears once in the feed, but checkpointing is your job. The processor gives at least once.
Save the token only after the batch is safely written downstream. If you crash in between, you read that batch again, so every downstream write must be idempotent.
Question 4: A teammate asks whether deletes appear in the change feed. What do you tell them?
Not in the default latest version mode: a deleted item just stops appearing, like order o2 earlier. All versions and deletes mode records every create, update and delete, with the intermediate versions too.
| Latest version (default) | All versions and deletes | |
|---|---|---|
| creates and updates | newest version only | every version, in order |
| deletes | not captured | captured: id, partition key, TTL flag |
| start from | beginning, time, now, token | now or a token only |
| requires | nothing: on by default | continuous backup (API for NoSQL) |
On Azure, it needs continuous backup on the account, and you can read back only as far as the backup retention:
# all versions and deletes: continuous backup, one-way
az cosmosdb update --name shop-prod --resource-group rg-shop \
--backup-policy-type Continuous \
--continuous-tier Continuous7DaysThe emulator wants a change feed policy on the container instead. Let's try it. avad.py reads from now to get a token, then creates, updates and deletes order o7. We run python avad.py.
from common import container, order
orders = container("orders_avad",
change_feed_policy={"retentionDuration": 10})
AVAD = "AllVersionsAndDeletes"
list(orders.query_items_change_feed(mode=AVAD, start_time="Now"))
tok = orders.client_connection.last_response_headers["etag"]
orders.create_item(order("o7", "c1", "new"))
orders.upsert_item(order("o7", "c1", "paid"))
orders.delete_item("o7", partition_key="c1")
for r in orders.query_items_change_feed(mode=AVAD, continuation=tok):
m = r["metadata"]
now = (r.get("current") or {}).get("status")
print(f"{m['operationType']:8} current {now}")
if m["operationType"] == "delete":
print(" delete of", m["id"], m["partitionKey"],
"ttl expired:", m["timeToLiveExpired"])$ python avad.py
create current new
replace current paid
delete current None
delete of o7 {'customerId': 'c1'} ttl expired: FalseThree rows come back: a create with current new, a replace with paid, and a delete with current None, so no current version. The delete carries the id, o7, the partition key, {'customerId': 'c1'}, and a flag saying whether time to live expired it. Here it's False.
The rules for this mode:
- Starts - from now or a token only
- API for NoSQL - only
- GA - since June 2026
Use a soft delete: set a deleted flag, let consumers remove the row, and let time to live purge the item later.
Question 5: Several workers must process the change feed in parallel. How does the change feed processor scale?
The processor splits the work by leases. Each lease covers one range of the container's partition keys. Instances with the same processor name and lease container form one deployment, and the leases are spread evenly over them. Add an instance, and leases move to it.
So instance A owns half the leases, and instance B owns the other half. The processor is in the .NET and Java SDKs. Here's the .NET shape: a processor name, an instance name, the lease container, and a delegate that handles each batch.
// .NET SDK V3 (not run: Python has no processor)
Container orders = client.GetContainer("shop", "orders");
Container leases = client.GetContainer("shop", "leases");
ChangeFeedProcessor processor = orders
.GetChangeFeedProcessorBuilder<Order>(
processorName: "orders-to-lake",
onChangesDelegate: HandleChangesAsync)
.WithInstanceName(Environment.MachineName)
.WithLeaseContainer(leases)
.Build();
await processor.StartAsync();Python has no processor, and Microsoft says use the pull model. So in Python, it's one worker per feed range, each with its own token file. workers.py lists the feed ranges and starts one worker per range in a thread pool. We run python workers.py twice.
import os
from concurrent.futures import ThreadPoolExecutor
from common import container
orders = container("orders")
ranges = list(orders.read_feed_ranges()) # per physical partition
print("feed ranges:", len(ranges))
def worker(n): # one worker per range
path = f"range{n}.token" # with its own position
tok = open(path).read() if os.path.exists(path) else None
kw = ({"continuation": tok} if tok else
{"feed_range": ranges[n], "start_time": "Beginning"})
pager = orders.query_items_change_feed(**kw).by_page()
ids = [d["id"] for page in pager for d in page]
if pager.continuation_token:
open(path, "w").write(pager.continuation_token)
return f"range {n}: {len(ids)} changes {ids}"
with ThreadPoolExecutor(max_workers=len(ranges)) as pool:
for line in pool.map(worker, range(len(ranges))):
print(line)$ python workers.py
feed ranges: 1
range 0: 5 changes ['o3', 'o4', 'o5', 'o6', 'o1']
$ python workers.py
feed ranges: 1
range 0: 0 changes []We get feed ranges: 1, and range 0 reads 5 changes. The second run reads 0 changes, because each worker resumed from its own token. The emulator has one physical partition, so one range. On Azure, there's one feed range per physical partition.
One limit: instances shouldn't outnumber leases. Extra instances just wait.
The processor delivers at least once. If your code throws, the same batch comes again, so keep the delegate idempotent.
Question 6: Someone wants an Azure Function to react to every new order. How does the Cosmos DB trigger work?
The Cosmos DB trigger reads the change feed for you. You name the database, the container, a connection setting and a lease container, and the function gets batches of new and updated documents. It doesn't get deletes.
We don't run a Functions host here, so this section is code. In Python, it's the Cosmos DB trigger decorator:
import logging
import azure.functions as func
app = func.FunctionApp()
@app.cosmos_db_trigger(arg_name="docs", database_name="shop",
container_name="orders",
connection="CosmosShop",
lease_container_name="leases")
def order_changed(docs: func.DocumentList):
for doc in docs: # new and updated orders
logging.info("order %s is %s", doc["id"], doc["status"])The lease container needs /id as its partition key. Here we create it with the Azure CLI:
# the lease container: partition key /id
az cosmosdb sql container create --account-name shop-prod \
--resource-group rg-shop --database-name shop \
--name leases --partition-key-path /idFor no secrets, use an identity-based connection: an account endpoint, credential set to managed identity, and a Cosmos DB data plane role for the identity, since the trigger writes leases. With an identity, don't let the trigger create the lease container, or the function won't start.
# Function App settings: identity-based, no key
CosmosShop__accountEndpoint=https://shop-prod.documents.azure.com
CosmosShop__credential=managedidentity
# + data plane role: Cosmos DB Built-in Data ContributorThen a few useful settings:
| Trigger setting | What it does |
|---|---|
| feedPollDelay | default 5000 ms between polls |
| startFromBeginning | first run only: no effect once leases exist |
| leaseContainerPrefix | share one lease container across functions |
One function per downstream system, each with its own lease prefix. Then a slow search update never holds back the lake.
Scenario 7: Every order change must flow to the data lake and a search index within a minute. How do you build it?
One reader, three steps per batch:
- Data lake - append raw changes (JSON lines)
- Search index - upsert by order id
- Checkpoint - token or lease, last
In Azure, the reader is a Functions trigger or pull workers per feed range. Here it's a real pull loop. A local JSON lines file stands in for the data lake, and a local file keyed by order id stands in for the search index.
Let's run it. The first poll backfills from the beginning. Then shop.py adds order o8 and pays o6, and we poll again. Last, we look at the end of the lake file. We run python pipeline.py, python shop.py, python pipeline.py and tail -n 2 lake/orders.jsonl.
import json, os
from common import container, biz
orders = container("orders")
TOKEN = "pipeline.token"
LAKE = "lake/orders.jsonl" # stands in for the data lake
INDEX = "search.json" # stands in for a search index
tok = open(TOKEN).read() if os.path.exists(TOKEN) else None
kw = {"continuation": tok} if tok else {"start_time": "Beginning"}
pager = orders.query_items_change_feed(**kw).by_page()
changes = [biz(d) for page in pager for d in page]
os.makedirs("lake", exist_ok=True)
with open(LAKE, "a") as lake: # 1. raw history
for d in changes:
lake.write(json.dumps(d, sort_keys=True) + "\n")
index = json.load(open(INDEX)) if os.path.exists(INDEX) else {}
for d in changes: # 2. upsert by id
index[d["id"]] = {"id": d["id"], "status": d["status"]}
json.dump(index, open(INDEX, "w"))
if pager.continuation_token: # 3. checkpoint last
open(TOKEN, "w").write(pager.continuation_token)
rows = sum(1 for _ in open(LAKE))
print(f"{len(changes)} changes | lake rows {rows}"
f" | index docs {len(index)}")from common import container, order
orders = container("orders")
orders.create_item(order("o8", "c1", "new"))
orders.upsert_item(order("o6", "c3", "paid"))
print("shop wrote: o8 new, o6 paid")$ python pipeline.py
5 changes | lake rows 5 | index docs 5
$ python shop.py
shop wrote: o8 new, o6 paid
$ python pipeline.py
2 changes | lake rows 7 | index docs 6
$ tail -n 2 lake/orders.jsonl
{"customerId": "c1", "id": "o8", "status": "new"}
{"customerId": "c3", "id": "o6", "status": "paid"}The first poll prints 5 changes | lake rows 5 | index docs 5. The next poll gets just the two new changes: lake rows 7 and index docs 6. The lake holds the latest version per poll, not every update, and the index holds one document per order. The last two lake rows are o8 with new and o6 with paid.
The checkpoint comes after both writes. A crash in between replays the batch: the lake tolerates duplicates, and the index upsert by id is idempotent.
| Concern | Answer |
|---|---|
| within a minute | trigger polls every 5 s by default |
| deletes | all versions and deletes, or soft delete |
| replays | duplicates ok in lake, upsert in index |
Within a minute is the easy part: the trigger polls every five seconds by default. The hard parts are deletes, which need all versions and deletes or a soft delete, and replays.
If the real need is reporting, consider Fabric mirroring, which is generally available: an analytics copy without your own pipeline.
Key points to remember
- The change feed: ordered per partition key, creates and updates, on by default
- Start from the beginning, a point in time, now, or a saved token
- Position = continuation token (pull) or leases (processor, Functions); checkpoint last
- Deletes: all versions and deletes mode (continuous backup) or a soft delete
- Scale with leases (.NET, Java) or one pull worker per feed range in Python
The full script
This is a Python script that writes every file and runs every command above in order against the Cosmos DB emulator: start the emulator, run pip install azure-cosmos first, set COSMOS_ENDPOINT and COSMOS_KEY, and run it in an empty folder on a freshly started emulator. A comment in it marks where to restart the emulator by hand.
Show the full script that writes the files and runs every step
# ========================================================================
# Cosmos DB Interview Q&A #7 - Change feed and event-driven pipelines
# TechBrothersIT. Names and sample data are invented for teaching.
# Every step below was run on the official Azure Cosmos DB Linux emulator (vNext, Docker image)
# with azure-cosmos 4.17.1; the outputs in the video and the blog post are the real outputs.
# To run it yourself:
# 1. docker run -p 8081:8081 -p 1234:1234 mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:vnext-latest
# and wait until its log says "Now listening on: http://0.0.0.0:8081".
# 2. pip install azure-cosmos
# 3. Set COSMOS_ENDPOINT=http://localhost:8081 and COSMOS_KEY to the emulator key
# (C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==),
# then run python CosmosDB_Interview_QA_07_Code.py. The scripts create their own databases and containers.
# Run it in an empty folder and on a freshly started emulator: it creates its files (and any local data
# files) there, and a second run on old data gives different output.
# 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('common.py', """\
import os
from azure.cosmos import CosmosClient, PartitionKey
client = CosmosClient(os.environ["COSMOS_ENDPOINT"],
credential=os.environ["COSMOS_KEY"])
db = client.create_database_if_not_exists("shop")
def container(name, **options):
return db.create_container_if_not_exists(
name, partition_key=PartitionKey(path="/customerId"),
**options)
def order(oid, customer, status):
return {"id": oid, "customerId": customer, "status": status}
def biz(item): # drop _rid, _etag, _ts, _lsn ...
return {k: v for k, v in item.items() if not k.startswith("_")}
""")
# Step 1: Six writes, then the change feed from the beginning: updates collapsed, delete missing
write('writes.py', """\
from common import container, order, biz
orders = container("orders")
orders.create_item(order("o1", "c1", "new"))
orders.create_item(order("o2", "c2", "new"))
orders.create_item(order("o3", "c1", "new"))
orders.upsert_item(order("o1", "c1", "paid"))
orders.upsert_item(order("o1", "c1", "shipped"))
orders.delete_item("o2", partition_key="c2")
print("6 writes: 3 creates, o1 updated twice, o2 deleted")
feed = list(orders.query_items_change_feed(start_time="Beginning"))
print("change feed from the beginning:", len(feed), "items")
for d in feed:
print(" ", biz(d))
print("o2 in the feed:", any(d["id"] == "o2" for d in feed))
""")
run('python writes.py')
# Step 2: Starting points: the beginning, a point in time, and now with its token
write('startpoints.py', """\
import time
from datetime import datetime, timezone
from common import container, order
orders = container("orders")
def ids(feed):
return [d["id"] for d in feed]
time.sleep(2)
mark = datetime.now(timezone.utc) # a point in time
time.sleep(2)
orders.create_item(order("o4", "c3", "new"))
print("from the beginning:",
ids(orders.query_items_change_feed(start_time="Beginning")))
print("from the point in time:",
ids(orders.query_items_change_feed(start_time=mark)))
now = orders.query_items_change_feed(start_time="Now")
list(now) # this read only fixes the position
tok = orders.client_connection.last_response_headers["etag"]
orders.create_item(order("o5", "c2", "new"))
print("from now, with its token:",
ids(orders.query_items_change_feed(continuation=tok)))
""")
run('python startpoints.py')
# Step 3: A consumer with a token file: first run from the beginning, then the shop writes
write('consumer.py', """\
import os
from common import container
orders = container("orders")
TOKEN = "orders.token" # the reader's position
tok = open(TOKEN).read() if os.path.exists(TOKEN) else None
print("start from:", "the saved token" if tok else "the beginning")
kw = {"continuation": tok} if tok else {"start_time": "Beginning"}
pager = orders.query_items_change_feed(**kw).by_page()
n = 0
for page in pager:
for d in page:
print(" ", d["id"], d["status"]) # process the change
n += 1
if pager.continuation_token: # None: nothing new
open(TOKEN, "w").write(pager.continuation_token)
print(f"changes: {n}, token saved")
else:
print(f"changes: {n}, nothing new")
""")
write('app.py', """\
from common import container, order
orders = container("orders")
orders.create_item(order("o6", "c3", "new"))
orders.upsert_item(order("o1", "c1", "delivered"))
print("shop wrote: o6 new, o1 delivered")
""")
run('python consumer.py')
run('python app.py')
# Step 4: After an emulator restart the consumer resumes from its saved token
# >>> now RESTART the emulator container (docker restart; the video restarts it here) <<<
run('python consumer.py')
run('python consumer.py')
# Step 5: All versions and deletes on a container with a change feed policy
write('avad.py', """\
from common import container, order
orders = container("orders_avad",
change_feed_policy={"retentionDuration": 10})
AVAD = "AllVersionsAndDeletes"
list(orders.query_items_change_feed(mode=AVAD, start_time="Now"))
tok = orders.client_connection.last_response_headers["etag"]
orders.create_item(order("o7", "c1", "new"))
orders.upsert_item(order("o7", "c1", "paid"))
orders.delete_item("o7", partition_key="c1")
for r in orders.query_items_change_feed(mode=AVAD, continuation=tok):
m = r["metadata"]
now = (r.get("current") or {}).get("status")
print(f"{m['operationType']:8} current {now}")
if m["operationType"] == "delete":
print(" delete of", m["id"], m["partitionKey"],
"ttl expired:", m["timeToLiveExpired"])
""")
run('python avad.py')
# Step 6: One pull-model worker per feed range, each with its own token
write('workers.py', """\
import os
from concurrent.futures import ThreadPoolExecutor
from common import container
orders = container("orders")
ranges = list(orders.read_feed_ranges()) # per physical partition
print("feed ranges:", len(ranges))
def worker(n): # one worker per range
path = f"range{n}.token" # with its own position
tok = open(path).read() if os.path.exists(path) else None
kw = ({"continuation": tok} if tok else
{"feed_range": ranges[n], "start_time": "Beginning"})
pager = orders.query_items_change_feed(**kw).by_page()
ids = [d["id"] for page in pager for d in page]
if pager.continuation_token:
open(path, "w").write(pager.continuation_token)
return f"range {n}: {len(ids)} changes {ids}"
with ThreadPoolExecutor(max_workers=len(ranges)) as pool:
for line in pool.map(worker, range(len(ranges))):
print(line)
""")
run('python workers.py')
run('python workers.py')
# Step 7: Orders to a local data lake file and search index: backfill, new changes, checkpoint last
write('pipeline.py', 'import json, os\nfrom common import container, biz\n\norders = container("orders")\nTOKEN = "pipeline.token"\nLAKE = "lake/orders.jsonl" # stands in for the data lake\nINDEX = "search.json" # stands in for a search index\ntok = open(TOKEN).read() if os.path.exists(TOKEN) else None\nkw = {"continuation": tok} if tok else {"start_time": "Beginning"}\npager = orders.query_items_change_feed(**kw).by_page()\nchanges = [biz(d) for page in pager for d in page]\n\nos.makedirs("lake", exist_ok=True)\nwith open(LAKE, "a") as lake: # 1. raw history\n for d in changes:\n lake.write(json.dumps(d, sort_keys=True) + "\\n")\nindex = json.load(open(INDEX)) if os.path.exists(INDEX) else {}\nfor d in changes: # 2. upsert by id\n index[d["id"]] = {"id": d["id"], "status": d["status"]}\njson.dump(index, open(INDEX, "w"))\nif pager.continuation_token: # 3. checkpoint last\n open(TOKEN, "w").write(pager.continuation_token)\nrows = sum(1 for _ in open(LAKE))\nprint(f"{len(changes)} changes | lake rows {rows}"\n f" | index docs {len(index)}")\n')
write('shop.py', """\
from common import container, order
orders = container("orders")
orders.create_item(order("o8", "c1", "new"))
orders.upsert_item(order("o6", "c3", "paid"))
print("shop wrote: o8 new, o6 paid")
""")
run('python pipeline.py')
run('python shop.py')
run('python pipeline.py')
run('tail -n 2 lake/orders.jsonl')
Part 8: Transactions, Concurrency and Server-Side Logic
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 Cosmos DB Linux emulator (vNext) with azure-cosmos 4.17.1. Features the emulator does not support are shown as code. Part of the Real Time Azure Cosmos DB 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 Cosmos DB documentation (learn.microsoft.com/azure/cosmos-db) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.