Cosmos DB Interview Questions and Answers Part 10 - Real-World Manager Scenarios

Cosmos DB Interview Q&A · Part 10 of 10

In this post we are going to cover real-world manager scenarios for 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 the final-round scenario: the hiring manager asks you to design the operational data platform for an online retailer on Cosmos DB.

This is the last part of the series. These questions aren't about one feature. A manager describes a problem, like throttling every morning, a slow query or numbers that don't match, and we walk through how we investigate it with data and then fix it. Interviewers ask them because they show whether you have run Cosmos DB on a real project.

Part 10 of the Real Time Azure Cosmos DB Interview Questions and Answers series on TechBrothersIT.

On this page

1. The short answer
2. Let's set up the project
3. Question 1: Requests are throttled every morning at nine. How do you investigate and fix it?
4. Question 2: A query that used to take milliseconds now takes seconds. Where do you look first?
5. Question 3: A bulk load of ten million documents is too slow. How do you speed it up?
6. Question 4: Data in Cosmos DB and the data warehouse don't match. How do you find the gap?
7. Question 5: A team wants to migrate from MongoDB to Cosmos DB. How do you plan it?
8. Question 6: A teammate asks how to test Cosmos DB code locally and in CI. What do you set up?
9. Scenario 7: In the final round, the hiring manager asks you to design the operational data platform for an online retailer on Cosmos DB. What is your design?
10. The full script

The short answer

Say this first

Investigate with data: RU consumption per partition key range and per key value for throttling, the partition key filter and query metrics for slow queries, the async client with bounded concurrency for loads, and an id + checksum reconciliation for gaps. Plan migrations with Azure DocumentDB tools, test on the emulator in CI, and design the platform from the access patterns.

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 one client per process from the COSMOS_ENDPOINT and COSMOS_KEY environment variables, and has two helpers: database creates a database, retail by default, and container creates a container partitioned by /customerId unless we pass another key.

common.py

import os
from azure.cosmos import CosmosClient, PartitionKey

URL = os.environ["COSMOS_ENDPOINT"]   # http://localhost:8081
KEY = os.environ["COSMOS_KEY"]        # the emulator's well-known key
client = CosmosClient(URL, credential=KEY)   # one client per process


def database(name="retail"):
    return client.create_database_if_not_exists(name)


def container(name, pk="/customerId", **options):
    return database().create_container_if_not_exists(
        name, partition_key=PartitionKey(path=pk), **options)

Question 1: Requests are throttled every morning at nine. How do you investigate and fix it?

Start with what changed at nine, then find the hot partition, then the key value behind it. Step by step:

• What changed at nine? - a deploy, a batch job, a campaign
• Insights: Normalized RU Consumption by PartitionKeyRangeID - one range at 100% = hot partition
• Diagnostic logs: CDBPartitionKeyRUConsumption - which key value; enable for a short time

Let's do the key value count locally. morning.py writes 600 orders in the nine o'clock rush, partitioned by /storeId, plus 60 in a quiet hour before. skew.py counts the nine o'clock writes per store ID. We run python morning.py and python skew.py.

morning.py

import common

orders = common.container("orders", pk="/storeId")
for i in range(600):                      # the 9 o'clock rush
    store = "online" if i % 3 else f"s{i // 3 % 10:02d}"
    orders.upsert_item({"id": f"o{i}", "storeId": store, "hour": 9,
                        "total": 10.5 + i % 40})
for i in range(60):                       # a quiet hour before
    orders.upsert_item({"id": f"e{i}", "storeId": f"s{i % 10:02d}",
                        "hour": 8, "total": 12.5})
print("orders written:", 660)

skew.py

import collections, common

orders = common.container("orders", pk="/storeId")
keys = orders.query_items(
    "SELECT VALUE c.storeId FROM c WHERE c.hour = 9",
    enable_cross_partition_query=True)
counts = collections.Counter(keys)
total = sum(counts.values())
print("writes from 9 to 10:", total, "| partition key values:",
      len(counts))
top = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))
for key, n in top[:4]:
    print(f"  storeId {key:7} {n:4} writes  {100 * n / total:5.1f} %")

Terminal

$ python morning.py
orders written: 660
$ python skew.py
writes from 9 to 10: 600 | partition key values: 11
  storeId online   400 writes   66.7 %
  storeId s00       20 writes    3.3 %
  storeId s01       20 writes    3.3 %
  storeId s02       20 writes    3.3 %

We get 600 writes over 11 partition key values, and online takes 400, 66.7 %. Every other store has 20. That's a hot key. Then the fix depends on the cause:

Cause	Fix
one hot key value	a better key: synthetic or hierarchical (new container)
a batch job at 9	priority-based execution: Low, or move the job
even load, too little RU/s	autoscale, or more RU/s

Meanwhile, the Python SDK retries 429s for you: 9 times or 30 seconds of waiting, honoring the server's retry-after. Only then do you get the error, so park that write and retry it later. The emulator never throttles, so this is code:

throttle.py

from azure.cosmos import CosmosClient
from azure.cosmos.exceptions import CosmosHttpResponseError

client = CosmosClient(URL, credential=cred,     # one per app
                      retry_throttle_total=9,         # default
                      retry_throttle_backoff_max=30)  # seconds
orders = client.get_database_client("retail") \
               .get_container_client("orders")
try:
    orders.upsert_item(order)   # 429s retried, honoring retry-after
except CosmosHttpResponseError as e:
    if e.status_code != 429:
        raise
    park_for_later(order)       # still throttled after the retries

# the 9 o'clock batch job: low priority (enabled on the account)
orders.upsert_item(batch_doc, priority="Low")

Metadata throttling

A new client per request causes it, and more throughput won't help. Use one client per app.

Question 2: A query that used to take milliseconds now takes seconds. Where do you look first?

First: did the query lose its partition key filter? That's the most common cause I see. Then the indexing policy, then the data:

• The partition key filter - lost in a refactor? Now every partition is read
• The indexing policy - an excluded path, a missing composite index
• The data - one key grew; far more documents read than returned

Let's reproduce the first one. slowseed.py writes 200 orders for 20 customers, partitioned by /customerId. After a refactor, the my orders page looks up by email instead of customer ID, so it's a cross-partition query. slow.py runs it both ways: across every partition, and with partition_key="c7". We run python slowseed.py and python slow.py.

slowseed.py

import common

orders = common.container("customer_orders")        # pk /customerId
for i in range(200):
    cust = f"c{i % 20}"
    orders.upsert_item({"id": f"o{i:03d}", "customerId": cust,
                        "email": f"{cust}@example.com",
                        "total": 10.25 + i % 30})
print("orders:", 200, "| customers:", 20)

slow.py

import common

orders = common.container("customer_orders")        # pk /customerId
sql = "SELECT c.id, c.total FROM c WHERE c.email = @e ORDER BY c.id"
p = [{"name": "@e", "value": "c7@example.com"}]
before = list(orders.query_items(sql, parameters=p,
                                 enable_cross_partition_query=True))
after = list(orders.query_items(sql, parameters=p,
                                partition_key="c7"))  # one partition
print("before, every partition:", len(before), "rows", before[:2])
print("after, partition_key=c7:", len(after), "rows", after[:2])
print("same rows:", before == after)

Terminal

$ python slowseed.py
orders: 200 | customers: 20
$ python slow.py
before, every partition: 10 rows [{'id': 'o007', 'total': 17.25}, {'id': 'o027', 'total': 37.25}]
after, partition_key=c7: 10 rows [{'id': 'o007', 'total': 17.25}, {'id': 'o027', 'total': 37.25}]
same rows: True

Both return 10 rows, starting with o007 and o027, and same rows: True. The emulator has one physical partition and charges 1 RU for everything, so the cost difference isn't visible here. On Azure:

On Azure (docs)	Request units
point read of 1 KB	1 RU
query with the partition key	one partition
query without it	+ 2-3 RU per physical partition

To see why a query is slow, turn on query metrics and index metrics. If it retrieves far more documents than it outputs, the index isn't doing the filtering.

metrics.py

rows = list(orders.query_items(sql, parameters=p,
    partition_key="c7", populate_query_metrics=True,
    populate_index_metrics=True))
h = orders.client_connection.last_response_headers
print(h["x-ms-request-charge"])            # RU, last page
print(h["x-ms-documentdb-query-metrics"])  # retrieved vs output
print(h["x-ms-cosmos-index-utilization"])  # index metrics

Real project

Log the request charge of your main queries, so a jump shows up on a dashboard before users notice.

Question 3: A bulk load of ten million documents is too slow. How do you speed it up?

Usually the loader writes one document at a time. The Python SDK has no bulk API, so the documented way is the async client with bounded concurrency.

Let's try it. In bulk.py, an asyncio.Semaphore(32) allows at most 32 writes at once, through one client. We send 3,000 orders. We run python bulk.py 3000.

bulk.py

import asyncio, os, sys
from azure.cosmos import PartitionKey
from azure.cosmos.aio import CosmosClient  # azure-cosmos[aio]

URL, KEY = os.environ["COSMOS_ENDPOINT"], os.environ["COSMOS_KEY"]

async def write(container, doc, limit):
    async with limit:                     # wait for a free slot
        await container.upsert_item(doc)

async def main(n):
    limit = asyncio.Semaphore(32)     # at most 32 writes in flight
    async with CosmosClient(URL, credential=KEY) as client:
        db = await client.create_database_if_not_exists("retail")
        co = await db.create_container_if_not_exists(
            "orders_load", partition_key=PartitionKey("/customerId"))
        docs = [{"id": f"o{i}", "customerId": f"c{i % 500}",
                 "total": 10.5 + i % 90} for i in range(n)]
        tasks = [write(co, d, limit) for d in docs]
        results = await asyncio.gather(*tasks,
                                       return_exceptions=True)
        failed = [r for r in results if isinstance(r, Exception)]
        count = [x async for x in co.query_items(
            "SELECT VALUE COUNT(1) FROM c")]
        print("sent:", n, "| failed:", len(failed),
              "| in the container:", count[0])

asyncio.run(main(int(sys.argv[1])))

Terminal

$ python bulk.py 3000
sent: 3000 | failed: 0 | in the container: 3000

sent: 3000 | failed: 0 | in the container: 3000. We print counts only, never timings.

Then make sure Cosmos DB can take it. A 1 KB insert without indexing costs around 5.5 RU, so 10 million documents need at least 55 million RU. Raise the autoscale maximum for the load, and lower it after, but never below a tenth of the highest maximum.

Then shape the load itself:

• Many key values - every partition works
• Index only what queries use
• Several workers - stream a slice each, in chunks

Indexing only the paths your queries need cuts the cost of every write.

Restartable

Use upserts with stable IDs, so running the load again creates no duplicates.

Question 4: Data in Cosmos DB and the data warehouse don't match. How do you find the gap?

Reconcile by key, then by content: read the ID and business fields from both sides, hash the fields into a checksum, and compare. Never include the ETag or the timestamp.

• Cosmos DB - container: sales
• Compare - id + checksum of business fields
• Warehouse - fact_orders (SQLite here)

Let's build it. seed.py loads 12 orders into the sales container and the same 12 into a SQLite fact_orders table: last night's load. today.py makes today's changes: one update, one patch, one delete and one new order. Then reconcile.py compares both sides. We run python seed.py, python today.py and python reconcile.py.

seed.py

import sqlite3, common

orders = common.container("sales")               # pk /customerId
rows = [(f"o{i:02d}", f"c{i % 4}", "paid", 15.5 + i)
        for i in range(1, 13)]
for oid, cust, status, total in rows:
    orders.upsert_item({"id": oid, "customerId": cust,
                        "status": status, "total": total})
wh = sqlite3.connect("warehouse.db")              # last night's load
wh.execute("CREATE TABLE fact_orders (order_id TEXT PRIMARY KEY,"
           " customer_id TEXT, status TEXT, total REAL)")
wh.executemany("INSERT INTO fact_orders VALUES (?, ?, ?, ?)", rows)
wh.commit()
print("cosmos and warehouse loaded:", len(rows), "orders each")

today.py

import common

orders = common.container("sales")
o3 = orders.read_item("o03", partition_key="c3")
orders.upsert_item(dict(o3, status="shipped"))       # an update
orders.patch_item("o07", partition_key="c3", patch_operations=[
    {"op": "set", "path": "/total", "value": 19.99}])  # a price fix
orders.delete_item("o05", partition_key="c1")        # a delete
orders.upsert_item({"id": "o13", "customerId": "c1",
                    "status": "paid", "total": 42.5})  # a new order
print("today: 1 update, 1 patch, 1 delete, 1 new order")

reconcile.py

import hashlib, json, sqlite3, common

def checksum(customer, status, total):   # business fields, no _etag
    text = json.dumps([customer, status, round(float(total), 2)])
    return hashlib.sha256(text.encode()).hexdigest()[:12]

orders = common.container("sales")
rows = orders.query_items("SELECT * FROM c",
                          enable_cross_partition_query=True)
cosmos = {r["id"]: checksum(r["customerId"], r["status"], r["total"])
          for r in rows}
wh = sqlite3.connect("warehouse.db")
dw = {oid: checksum(cust, status, total) for oid, cust, status, total
      in wh.execute("SELECT * FROM fact_orders")}
print("rows | cosmos:", len(cosmos), "| warehouse:", len(dw))
print("missing in warehouse:", sorted(cosmos.keys() - dw.keys()))
print("only in warehouse:   ", sorted(dw.keys() - cosmos.keys()))
print("changed:", sorted(k for k in cosmos.keys() & dw.keys()
                         if cosmos[k] != dw[k]))

Terminal

$ python seed.py
cosmos and warehouse loaded: 12 orders each
$ python today.py
today: 1 update, 1 patch, 1 delete, 1 new order
$ python reconcile.py
rows | cosmos: 12 | warehouse: 12
missing in warehouse: ['o13']
only in warehouse:    ['o05']
changed: ['o03', 'o07']

Both sides have 12 rows, so a count alone looks fine. But the reconciliation finds all four: o13 is missing in the warehouse, o05 is only in the warehouse, because it was deleted in Cosmos DB, and o03 and o07 changed. The checksum rounds the total, so number types don't raise false alarms.

Then find out why:

Gap	Usual cause
only in the warehouse	a delete or TTL expiry: not in the latest version change feed
missing in the warehouse	a late pipeline, or a checkpoint saved too early
changed	an update the pipeline skipped

The latest version change feed has no deletes, so a change feed pipeline never sees order 5. Use all versions and deletes mode, which needs continuous backup, or soft deletes. Time to live expiry is a delete too.

Check the checkpoint

Saving the continuation token before the write succeeds skips changes after a crash. Reconcile daily and alert on any gap.

Question 5: A team wants to migrate from MongoDB to Cosmos DB. How do you plan it?

First pick the target. To keep MongoDB drivers and tools, it's Azure DocumentDB, formerly vCore-based Azure Cosmos DB for MongoDB. The API for NoSQL is a different API, so you redesign the data model and the code.

Target	What changes	Tooling
Azure DocumentDB (formerly vCore-based Azure Cosmos DB for MongoDB)	little: MongoDB drivers and tools keep working	VS Code migration extension, Database Migration Service
Azure Cosmos DB for NoSQL	a cross-API move: new data model, new code	your own copy, Desktop Data Migration Tool

For the copy, Microsoft lists the Azure DocumentDB migration extension for VS Code, designed for zero downtime, and the Azure Database Migration Service, online or offline. mongodump and mongorestore, or mongoexport and mongoimport, are offline only.

Tool	Mode	Notes
Azure DocumentDB Migration Extension for VS Code	designed for zero downtime	also runs the premigration assessment
Azure Database Migration Service	online or offline	runs in Azure
mongodump / mongorestore, mongoexport / mongoimport	offline	native MongoDB tools

Then plan in phases:

• Assess - the premigration assessment
• Trial migration - before the real one
• Online prerequisites - change streams, a large enough oplog
• TTL indexes - drop during the migration, recreate after
• Validate - counts and checksums per collection
• Cutover - no rollback: keep the source

Cutover has no rollback, so keep the source until you're sure.

A small trap

URL-encode the password in the connection string. Special characters break it.

Question 6: A teammate asks how to test Cosmos DB code locally and in CI. What do you set up?

The Linux emulator: Microsoft's official Docker image, tag vnext-latest, generally available, no Azure subscription needed.

Emulator port	What it serves
8081	the gateway endpoint: http://localhost:8081 (HTTP by default)
1234	Data Explorer
8080	health probes: /alive, /ready, /status

Locally, docker compose starts it next to the app:

docker-compose.yml

# docker-compose.yml: the emulator next to the app
services:
  cosmos:
    image: mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:vnext-latest
    ports:
      - "8081:8081"    # gateway endpoint (HTTP)
      - "1234:1234"    # Data Explorer
      - "8080:8080"    # health: /alive /ready /status

In GitHub Actions it runs as a service container, and the job waits for the ready check before testing:

test.yml

# .github/workflows/test.yml
name: test
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    services:
      cosmos:
        image: mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:vnext-latest
        ports: ["8081:8081", "8080:8080"]
    env:
      COSMOS_ENDPOINT: http://localhost:8081
      COSMOS_KEY: ${{ vars.COSMOS_EMULATOR_KEY }}  # well-known key
    steps:
      - uses: actions/checkout@v5
      - uses: actions/setup-python@v6
        with:
          python-version: "3.13"
      - run: pip install azure-cosmos
      - name: wait for the emulator
        timeout-minutes: 3
        run: until curl -sf localhost:8080/ready; do sleep 2; done
      - run: python run_tests.py

The tests are plain functions with asserts. Each run uses a fresh container with a random name, so runs never share data. Pytest isn't installed here, so run_tests.py calls the tests, deletes the container, and sets the exit code that fails the build. We run python run_tests.py.

test_orders.py

import uuid, common
from azure.core import MatchConditions
from azure.cosmos import exceptions

orders = common.container("test_" + uuid.uuid4().hex[:8])  # fresh

def test_create_then_read():
    orders.create_item({"id": "t1", "customerId": "c1", "n": 1.5})
    assert orders.read_item("t1", partition_key="c1")["n"] == 1.5

def test_duplicate_id_is_409():
    orders.create_item({"id": "t2", "customerId": "c1"})
    try:
        orders.create_item({"id": "t2", "customerId": "c1"})
        assert False, "expected a conflict"
    except exceptions.CosmosResourceExistsError as e:
        assert e.status_code == 409

def test_stale_etag_is_412():
    doc = orders.create_item({"id": "t3", "customerId": "c1", "n": 1})
    orders.replace_item("t3", dict(doc, n=2))     # someone else wins
    try:
        orders.replace_item("t3", dict(doc, n=3), etag=doc["_etag"],
            match_condition=MatchConditions.IfNotModified)
        assert False, "expected a precondition failure"
    except exceptions.CosmosAccessConditionFailedError as e:
        assert e.status_code == 412

run_tests.py

import sys, common
import test_orders as suite            # pytest isn't installed here

failed = 0
tests = [n for n in dir(suite) if n.startswith("test_")]
for name in tests:
    try:
        getattr(suite, name)()
        print("PASS", name)
    except Exception as e:
        failed += 1
        print("FAIL", name, type(e).__name__, e)
common.database().delete_container(suite.orders)      # clean up
print(len(tests) - failed, "passed,", failed, "failed")
sys.exit(1 if failed else 0)

Terminal

$ python run_tests.py
PASS test_create_then_read
PASS test_duplicate_id_is_409
PASS test_stale_etag_is_412
3 passed, 0 failed

All three pass: create and read, a duplicate ID gives 409, and a stale ETag gives 412. The last line is 3 passed, 0 failed.

Know the limits

The emulator has no real request units, no throttling, no indexing policy effect and no key checks. Keep a smoke test on a real dev account too.

Scenario 7: In the final round, the hiring manager asks you to design the operational data platform for an online retailer on Cosmos DB. What is your design?

One container per access pattern:

• catalog - /categoryId
• carts - /customerId, TTL 30 days
• orders - /customerId, lines not indexed

Carts get a 30-day time to live for abandoned carts. Orders leave the order lines out of the index, since nobody filters on them.

Let's create them. platform_setup.py creates the three containers from one plan and reads each one back. Then flow.py writes two orders, ships order o1 after it was paid, and reads the change feed from the beginning into a search index file. We run python platform_setup.py and python flow.py.

platform_setup.py

import common
from azure.cosmos import PartitionKey

DAY = 24 * 3600
db = common.database("platform")
PLAN = [("catalog", "/categoryId", {}),
        ("carts", "/customerId", {"default_ttl": 30 * DAY}),
        ("orders", "/customerId", {"indexing_policy": {
            "includedPaths": [{"path": "/*"}],
            "excludedPaths": [{"path": "/lines/*"}]}})]
for name, pk, options in PLAN:
    props = db.create_container_if_not_exists(
        name, partition_key=PartitionKey(path=pk), **options).read()
    excluded = [p["path"] for p in props["indexingPolicy"]
                ["excludedPaths"] if not p["path"].startswith('/"_')]
    print(f"{name:8} key {props['partitionKey']['paths'][0]:12}",
          "| ttl", props.get("defaultTtl"), "| not indexed", excluded)

flow.py

import json, common

orders = common.database("platform").get_container_client("orders")
orders.upsert_item({"id": "o1", "customerId": "c1", "status": "paid",
                    "lines": [{"sku": "p7", "qty": 2}]})
orders.upsert_item({"id": "o2", "customerId": "c2", "status": "paid",
                    "lines": [{"sku": "p3", "qty": 1}]})
orders.patch_item("o1", partition_key="c1", patch_operations=[
    {"op": "set", "path": "/status", "value": "shipped"}])
feed = orders.query_items_change_feed(start_time="Beginning")
with open("search_index.jsonl", "w") as out:   # stands in for search
    for change in feed:
        doc = {"id": change["id"], "status": change["status"]}
        out.write(json.dumps(doc) + "\n")
        print("change feed ->", doc)

Terminal

$ python platform_setup.py
catalog  key /categoryId  | ttl None | not indexed []
carts    key /customerId  | ttl 2592000 | not indexed []
orders   key /customerId  | ttl None | not indexed ['/lines/*']
$ python flow.py
change feed -> {'id': 'o2', 'status': 'paid'}
change feed -> {'id': 'o1', 'status': 'shipped'}

The plan reads back: catalog on /categoryId, carts with ttl 2592000, which is 30 days in seconds, and orders with ['/lines/*'] not indexed. The feed delivers only the latest version of each order: o2 as paid and o1 as shipped. This is the pull model in Python. In production, you save the continuation token.

Then the rest of the platform around it:

• Reports - Fabric mirroring: no app RUs used
• Throughput - autoscale per container
• Consistency - Session, the default
• Regions - read regions, service-managed failover
• Backup - continuous, point-in-time restore
• Security - Entra ID, private endpoints, all in Bicep

Reports go through Fabric mirroring, so analysts never touch the app's request units.

Name the trade-offs

Transactions stay inside one logical partition, deletes need all versions and deletes mode downstream, and everything is tested on the emulator in CI, with load tests on Azure.

Key points to remember

• Throttling: RU by PartitionKeyRangeID, then counts per key value; fix the key, the job or the RU/s
• Slow query: check the partition key filter first, then indexing and data, with query metrics
• Bulk load: async client + bounded concurrency, enough RU/s, stable IDs
• Gaps: reconcile by id + checksum; deletes need all versions and deletes mode
• MongoDB: Azure DocumentDB tools; test on the emulator in CI

The full script

This is a Python script that writes every file and runs every command above in order against the Cosmos DB emulator: run pip install azure-cosmos first (the bulk load also needs pip install aiohttp), set COSMOS_ENDPOINT and COSMOS_KEY, and run it in an empty folder on a freshly started emulator.

Show the full script that writes the files and runs every step

Full script - writes the files and runs every step against the Cosmos DB emulator (pip install azure-cosmos), run it with Python

# ========================================================================
# Cosmos DB Interview Q&A #10 - Real-world manager scenarios
# 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   (scripts that use azure.cosmos.aio also need: pip install aiohttp)
#   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_10_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

URL = os.environ["COSMOS_ENDPOINT"]   # http://localhost:8081
KEY = os.environ["COSMOS_KEY"]        # the emulator's well-known key
client = CosmosClient(URL, credential=KEY)   # one client per process


def database(name="retail"):
    return client.create_database_if_not_exists(name)


def container(name, pk="/customerId", **options):
    return database().create_container_if_not_exists(
        name, partition_key=PartitionKey(path=pk), **options)
""")

# Step 1: The 9 o'clock rush: 600 writes, then count writes per partition key value
write('morning.py', """\
import common

orders = common.container("orders", pk="/storeId")
for i in range(600):                      # the 9 o'clock rush
    store = "online" if i % 3 else f"s{i // 3 % 10:02d}"
    orders.upsert_item({"id": f"o{i}", "storeId": store, "hour": 9,
                        "total": 10.5 + i % 40})
for i in range(60):                       # a quiet hour before
    orders.upsert_item({"id": f"e{i}", "storeId": f"s{i % 10:02d}",
                        "hour": 8, "total": 12.5})
print("orders written:", 660)
""")
write('skew.py', """\
import collections, common

orders = common.container("orders", pk="/storeId")
keys = orders.query_items(
    "SELECT VALUE c.storeId FROM c WHERE c.hour = 9",
    enable_cross_partition_query=True)
counts = collections.Counter(keys)
total = sum(counts.values())
print("writes from 9 to 10:", total, "| partition key values:",
      len(counts))
top = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))
for key, n in top[:4]:
    print(f"  storeId {key:7} {n:4} writes  {100 * n / total:5.1f} %")
""")
run('python morning.py')
run('python skew.py')

# Step 2: The same query cross-partition and with partition_key=: same rows
write('slowseed.py', """\
import common

orders = common.container("customer_orders")        # pk /customerId
for i in range(200):
    cust = f"c{i % 20}"
    orders.upsert_item({"id": f"o{i:03d}", "customerId": cust,
                        "email": f"{cust}@example.com",
                        "total": 10.25 + i % 30})
print("orders:", 200, "| customers:", 20)
""")
write('slow.py', """\
import common

orders = common.container("customer_orders")        # pk /customerId
sql = "SELECT c.id, c.total FROM c WHERE c.email = @e ORDER BY c.id"
p = [{"name": "@e", "value": "c7@example.com"}]
before = list(orders.query_items(sql, parameters=p,
                                 enable_cross_partition_query=True))
after = list(orders.query_items(sql, parameters=p,
                                partition_key="c7"))  # one partition
print("before, every partition:", len(before), "rows", before[:2])
print("after, partition_key=c7:", len(after), "rows", after[:2])
print("same rows:", before == after)
""")
run('python slowseed.py')
run('python slow.py')

# Step 3: A bulk load of 3,000 orders with the async client and bounded concurrency
write('bulk.py', """\
import asyncio, os, sys
from azure.cosmos import PartitionKey
from azure.cosmos.aio import CosmosClient  # azure-cosmos[aio]

URL, KEY = os.environ["COSMOS_ENDPOINT"], os.environ["COSMOS_KEY"]

async def write(container, doc, limit):
    async with limit:                     # wait for a free slot
        await container.upsert_item(doc)

async def main(n):
    limit = asyncio.Semaphore(32)     # at most 32 writes in flight
    async with CosmosClient(URL, credential=KEY) as client:
        db = await client.create_database_if_not_exists("retail")
        co = await db.create_container_if_not_exists(
            "orders_load", partition_key=PartitionKey("/customerId"))
        docs = [{"id": f"o{i}", "customerId": f"c{i % 500}",
                 "total": 10.5 + i % 90} for i in range(n)]
        tasks = [write(co, d, limit) for d in docs]
        results = await asyncio.gather(*tasks,
                                       return_exceptions=True)
        failed = [r for r in results if isinstance(r, Exception)]
        count = [x async for x in co.query_items(
            "SELECT VALUE COUNT(1) FROM c")]
        print("sent:", n, "| failed:", len(failed),
              "| in the container:", count[0])

asyncio.run(main(int(sys.argv[1])))
""")
run('python bulk.py 3000')

# Step 4: Reconcile Cosmos DB with a warehouse (SQLite) by id and a checksum of business fields
write('seed.py', """\
import sqlite3, common

orders = common.container("sales")               # pk /customerId
rows = [(f"o{i:02d}", f"c{i % 4}", "paid", 15.5 + i)
        for i in range(1, 13)]
for oid, cust, status, total in rows:
    orders.upsert_item({"id": oid, "customerId": cust,
                        "status": status, "total": total})
wh = sqlite3.connect("warehouse.db")              # last night's load
wh.execute("CREATE TABLE fact_orders (order_id TEXT PRIMARY KEY,"
           " customer_id TEXT, status TEXT, total REAL)")
wh.executemany("INSERT INTO fact_orders VALUES (?, ?, ?, ?)", rows)
wh.commit()
print("cosmos and warehouse loaded:", len(rows), "orders each")
""")
write('today.py', """\
import common

orders = common.container("sales")
o3 = orders.read_item("o03", partition_key="c3")
orders.upsert_item(dict(o3, status="shipped"))       # an update
orders.patch_item("o07", partition_key="c3", patch_operations=[
    {"op": "set", "path": "/total", "value": 19.99}])  # a price fix
orders.delete_item("o05", partition_key="c1")        # a delete
orders.upsert_item({"id": "o13", "customerId": "c1",
                    "status": "paid", "total": 42.5})  # a new order
print("today: 1 update, 1 patch, 1 delete, 1 new order")
""")
write('reconcile.py', """\
import hashlib, json, sqlite3, common

def checksum(customer, status, total):   # business fields, no _etag
    text = json.dumps([customer, status, round(float(total), 2)])
    return hashlib.sha256(text.encode()).hexdigest()[:12]

orders = common.container("sales")
rows = orders.query_items("SELECT * FROM c",
                          enable_cross_partition_query=True)
cosmos = {r["id"]: checksum(r["customerId"], r["status"], r["total"])
          for r in rows}
wh = sqlite3.connect("warehouse.db")
dw = {oid: checksum(cust, status, total) for oid, cust, status, total
      in wh.execute("SELECT * FROM fact_orders")}
print("rows | cosmos:", len(cosmos), "| warehouse:", len(dw))
print("missing in warehouse:", sorted(cosmos.keys() - dw.keys()))
print("only in warehouse:   ", sorted(dw.keys() - cosmos.keys()))
print("changed:", sorted(k for k in cosmos.keys() & dw.keys()
                         if cosmos[k] != dw[k]))
""")
run('python seed.py')
run('python today.py')
run('python reconcile.py')

# Step 5: Tests on the emulator: a fresh container per run
write('test_orders.py', """\
import uuid, common
from azure.core import MatchConditions
from azure.cosmos import exceptions

orders = common.container("test_" + uuid.uuid4().hex[:8])  # fresh

def test_create_then_read():
    orders.create_item({"id": "t1", "customerId": "c1", "n": 1.5})
    assert orders.read_item("t1", partition_key="c1")["n"] == 1.5

def test_duplicate_id_is_409():
    orders.create_item({"id": "t2", "customerId": "c1"})
    try:
        orders.create_item({"id": "t2", "customerId": "c1"})
        assert False, "expected a conflict"
    except exceptions.CosmosResourceExistsError as e:
        assert e.status_code == 409

def test_stale_etag_is_412():
    doc = orders.create_item({"id": "t3", "customerId": "c1", "n": 1})
    orders.replace_item("t3", dict(doc, n=2))     # someone else wins
    try:
        orders.replace_item("t3", dict(doc, n=3), etag=doc["_etag"],
            match_condition=MatchConditions.IfNotModified)
        assert False, "expected a precondition failure"
    except exceptions.CosmosAccessConditionFailedError as e:
        assert e.status_code == 412
""")
write('run_tests.py', """\
import sys, common
import test_orders as suite            # pytest isn't installed here

failed = 0
tests = [n for n in dir(suite) if n.startswith("test_")]
for name in tests:
    try:
        getattr(suite, name)()
        print("PASS", name)
    except Exception as e:
        failed += 1
        print("FAIL", name, type(e).__name__, e)
common.database().delete_container(suite.orders)      # clean up
print(len(tests) - failed, "passed,", failed, "failed")
sys.exit(1 if failed else 0)
""")
run('python run_tests.py')

# Step 6: The platform containers from one plan, read back; the change feed to a search index file
write('platform_setup.py', """\
import common
from azure.cosmos import PartitionKey

DAY = 24 * 3600
db = common.database("platform")
PLAN = [("catalog", "/categoryId", {}),
        ("carts", "/customerId", {"default_ttl": 30 * DAY}),
        ("orders", "/customerId", {"indexing_policy": {
            "includedPaths": [{"path": "/*"}],
            "excludedPaths": [{"path": "/lines/*"}]}})]
for name, pk, options in PLAN:
    props = db.create_container_if_not_exists(
        name, partition_key=PartitionKey(path=pk), **options).read()
    excluded = [p["path"] for p in props["indexingPolicy"]
                ["excludedPaths"] if not p["path"].startswith('/"_')]
    print(f"{name:8} key {props['partitionKey']['paths'][0]:12}",
          "| ttl", props.get("defaultTtl"), "| not indexed", excluded)
""")
write('flow.py', 'import json, common\n\norders = common.database("platform").get_container_client("orders")\norders.upsert_item({"id": "o1", "customerId": "c1", "status": "paid",\n                    "lines": [{"sku": "p7", "qty": 2}]})\norders.upsert_item({"id": "o2", "customerId": "c2", "status": "paid",\n                    "lines": [{"sku": "p3", "qty": 1}]})\norders.patch_item("o1", partition_key="c1", patch_operations=[\n    {"op": "set", "path": "/status", "value": "shipped"}])\nfeed = orders.query_items_change_feed(start_time="Beginning")\nwith open("search_index.jsonl", "w") as out:   # stands in for search\n    for change in feed:\n        doc = {"id": change["id"], "status": change["status"]}\n        out.write(json.dumps(doc) + "\\n")\n        print("change feed ->", doc)\n')
run('python platform_setup.py')
run('python flow.py')

That is the whole series

All 10 parts are listed on the Real Time Azure Cosmos DB Interview Questions and Answers page on this blog.

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.