In this post we are going to cover real-world manager scenarios for Python data engineers: a file loaded twice into the CFO's report, scripts versus Airflow or dbt, a memory leak, a new source by Friday, a codebase with no tests, a vendor that changes its file format, and a design for fifty sources. These are 7 real Python interview questions, asked the way they come up on a real project, and the last one is a scenario from the final round: the hiring manager asks you to design a Python ingestion framework for fifty sources.
This is the last post in the series of seventy real-time questions, from core Python and pandas to APIs, databases, testing and real manager scenarios. These questions don't have one right line of code. The interviewer wants to hear how you think: who you tell first, how you find the scope, how you fix it safely, and what you put in place so it doesn't happen again. We still back every answer with real Python.
Every Python step below runs on the sample data, and the output under each step is the real result.
Part 10 of the Real Time Python for Data Engineers Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's create some sample data
- Question 1: The CFO's report is wrong because a Python job loaded yesterday's file twice. Walk me through your response.
- Question 2: Leadership asks whether to keep Python scripts or move the pipelines to a tool like Airflow or dbt. How do you decide?
- Question 3: A Python job uses more memory every night until it crashes. How do you find and fix the leak?
- Question 4: The business wants a new data source loaded by Friday. How do you scope and deliver it safely?
- Question 5: Your team's Python code has no tests and breaks often. What do you put in place first?
- Question 6: A vendor changes their file format without notice and the load fails. How do you make the pipeline handle it?
- Scenario 7: In the final round, the hiring manager asks you to design a Python ingestion framework for fifty sources. What is your design?
- The full script
The short answer
In an incident, communicate first, find the scope, fix the data in one transaction, and make the load safe to rerun. Choose tools by the pain, not the hype, find leaks with tracemalloc, scope deadlines to a safe first slice, add guardrails before coverage, check every file against a contract, and build one config-driven framework instead of fifty scripts.
Keep that in mind. Every question below is one piece of it.
Let's create some sample data
The setup builds a small SQLite warehouse, warehouse.db, where the October 22 sales file was loaded twice, plus a staging table with today's file. It also writes a leaky nightly job and its fixed version, a few small CSV files for the other questions, and a sources.toml config with fake connectors that stand in for a real API and database.
import sqlite3
from pathlib import Path
from typing import Protocol
# Q1: the warehouse table after yesterday's file was loaded twice
con = sqlite3.connect("warehouse.db", autocommit=False)
con.executescript("""
CREATE TABLE fact_sales (
sale_id INTEGER NOT NULL,
sale_date TEXT NOT NULL,
amount INTEGER NOT NULL,
source_file TEXT NOT NULL,
load_id INTEGER NOT NULL
);
-- load 41 and load 42 both loaded the Oct 22 file
INSERT INTO fact_sales VALUES
(5001, '2026-10-21', 120, 'sales_2026-10-21.csv', 40),
(5002, '2026-10-21', 80, 'sales_2026-10-21.csv', 40),
(5003, '2026-10-21', 100, 'sales_2026-10-21.csv', 40),
(5004, '2026-10-22', 150, 'sales_2026-10-22.csv', 41),
(5005, '2026-10-22', 50, 'sales_2026-10-22.csv', 41),
(5006, '2026-10-22', 40, 'sales_2026-10-22.csv', 41),
(5004, '2026-10-22', 150, 'sales_2026-10-22.csv', 42),
(5005, '2026-10-22', 50, 'sales_2026-10-22.csv', 42),
(5006, '2026-10-22', 40, 'sales_2026-10-22.csv', 42);
-- staging: today's file, as the job landed it
CREATE TABLE stg_sales (sale_id INTEGER, sale_date TEXT, amount INTEGER, source_file TEXT);
INSERT INTO stg_sales VALUES
(5007, '2026-10-23', 200, 'sales_2026-10-23.csv'),
(5008, '2026-10-23', 90, 'sales_2026-10-23.csv'),
(5009, '2026-10-23', 60, 'sales_2026-10-23.csv');
""")
con.commit()
UPSERT_SQL = """
INSERT INTO fact_sales
(sale_id, sale_date, amount, source_file, load_id)
SELECT sale_id, sale_date, amount, source_file, :load_id
FROM stg_sales
WHERE true -- SQLite: a WHERE is needed before ON CONFLICT
ON CONFLICT (sale_id) DO UPDATE SET
amount = excluded.amount,
load_id = excluded.load_id"""
# Q3: the leaky nightly job, and the fixed one
Path("job.py").write_text('''\
_seen: dict[int, dict] = {} # module level: lives forever
def run(night: int) -> int:
"""One nightly run inside a long-lived worker."""
for i in range(5_000):
order_id = night * 5_000 + i
_seen[order_id] = {"id": order_id, "status": "open"}
return len(_seen)
''', encoding="utf-8")
Path("job_fixed.py").write_text('''\
def run(night: int) -> int:
"""One run: the cache is freed when it returns."""
seen: dict[int, dict] = {}
for i in range(5_000):
order_id = night * 5_000 + i
seen[order_id] = {"id": order_id, "status": "open"}
return len(seen)
''', encoding="utf-8")
# Q4, Q5, Q6: sample files
Path("returns_sample.csv").write_text("""\
return_id,store_id,return_date,amount,reason
9001,12,2026-10-01,25.00,damaged
9002,12,2026-10-01,40.00,
9003,15,2026-10-02,15.50,wrong size
9003,15,2026-10-02,15.50,wrong size
9004,18,2026-10-03,60.00,
9005,12,2026-10-04,22.00,changed mind
9006,15,2026-10-05,35.00,damaged
9007,18,2026-10-06,18.00,wrong size
""", encoding="utf-8")
Path("orders_today.csv").write_text("""\
order_id,amount
1001,120.00
1002,
1003,80.00
1003,80.00
1004,
1005,45.00
""", encoding="utf-8")
Path("orders_2026-10-22.csv").write_text("""\
order_id,order_date,amount
1001,2026-10-22,120.00
1002,2026-10-22,75.50
""", encoding="utf-8")
Path("orders_2026-10-23.csv").write_text("""\
order_id,order_date,channel,total_amount
ORD-1003,2026-10-23,web,64.00
ORD-1004,2026-10-23,store,210.00
""", encoding="utf-8")
# Q7: the source config, and connectors (fakes stand in for the real API and database)
Path("sources.toml").write_text('''\
[sources.orders_api]
kind = "api"
key = "order_id"
mode = "incremental"
options = { endpoint = "/v1/orders" }
[sources.erp_customers]
kind = "db"
key = "customer_id"
mode = "full"
options = { table = "dbo.customers" }
''', encoding="utf-8")
class Connector(Protocol):
def extract(self) -> list[dict]: ...
class FakeApi: # stands in for a REST API connector
def __init__(self, endpoint: str) -> None:
self.endpoint = endpoint
def extract(self) -> list[dict]:
return [{"order_id": n} for n in (1001, 1002, 1003)]
class FakeDb: # stands in for a database connector
def __init__(self, table: str) -> None:
self.table = table
def extract(self) -> list[dict]:
return [{"customer_id": n} for n in (21, None, 23, 24)]
CONNECTORS: dict[str, type[Connector]] = {"api": FakeApi, "db": FakeDb}Question 1: The CFO's report is wrong because a Python job loaded yesterday's file twice. Walk me through your response.
The people first, then the data. Here's the order I follow:
- Tell the CFO's team: what is wrong, and when they'll hear next.
- Stop the damage: pause the job, and hold any report refresh that reads the table.
- Find the bad load: rows per load ID.
- Fix it in one transaction.
- Make the load safe to rerun, then a post-mortem.
Every row carries a load ID, so one query shows the scope. Let's count rows and totals per file and load.
BY_LOAD = """
SELECT source_file, load_id, COUNT(*), SUM(amount)
FROM fact_sales
GROUP BY source_file, load_id
ORDER BY load_id"""
for file, load_id, rows, total in con.execute(BY_LOAD):
print(f"{file} load {load_id}: {rows} rows, {total}")sales_2026-10-21.csv load 40: 3 rows, 300
sales_2026-10-22.csv load 41: 3 rows, 240
sales_2026-10-22.csv load 42: 3 rows, 240Yesterday's file, sales_2026-10-22.csv, came in with load 41 and again with load 42: 3 rows and 240 each time.
Now let's fix it in one transaction: delete load 42, and add a unique index on the sale ID, so the same sale can't be stored twice.
with con: # one transaction: commit, or roll back on error
cur = con.execute(
"DELETE FROM fact_sales WHERE load_id = ?", (42,))
con.execute("CREATE UNIQUE INDEX ux_fact_sales_sale_id"
" ON fact_sales (sale_id)")
print("deleted rows:", cur.rowcount)
for file, load_id, rows, total in con.execute(BY_LOAD):
print(f"{file} load {load_id}: {rows} rows, {total}")deleted rows: 3
sales_2026-10-21.csv load 40: 3 rows, 300
sales_2026-10-22.csv load 41: 3 rows, 240deleted rows: 3, and each file now appears once.
Then make the load safe to rerun, which matters most in warehouses that don't enforce keys. It upserts from staging on the sale ID:
INSERT INTO fact_sales
(sale_id, sale_date, amount, source_file, load_id)
SELECT sale_id, sale_date, amount, source_file, :load_id
FROM stg_sales
WHERE true -- SQLite: a WHERE is needed before ON CONFLICT
ON CONFLICT (sale_id) DO UPDATE SET
amount = excluded.amount,
load_id = excluded.load_idLet's load today's file twice.
DAY_TOTAL = """
SELECT COUNT(*), SUM(amount) FROM fact_sales
WHERE sale_date = ?"""
day = "2026-10-23"
for load_id in (43, 44): # the same file, loaded twice
with con:
con.execute(UPSERT_SQL, {"load_id": load_id})
n, total = con.execute(DAY_TOTAL, (day,)).fetchone()
print(f"after load {load_id}: {n} rows, total {total}")
con.close() # with con: commits or rolls back, never closesafter load 43: 3 rows, total 350
after load 44: 3 rows, total 3503 rows, total 350 after load 43, and the same after load 44.
Keep it blameless: why did it run twice, and why did no check catch it? Add row count and total checks against the source, and confirm the fixed numbers with the CFO's team.
Question 2: Leadership asks whether to keep Python scripts or move the pipelines to a tool like Airflow or dbt. How do you decide?
I start with the pain, not the tool. I ask:
- How many jobs?
- Do they depend on each other?
- What breaks? Retries, reruns, alerts.
- Few and independent? Scripts are fine, if they log, alert and rerun safely.
Once jobs depend on each other, you're writing an orchestrator yourself. Python's graphlib can order tasks by their dependencies. Let's try it on six tasks, from the two extracts to the report refresh.
from graphlib import TopologicalSorter
deps = {"load_sales": ["extract_sales"],
"load_customers": ["extract_customers"],
"build_gold": ["load_sales", "load_customers"],
"refresh_report": ["build_gold"]}
graph = TopologicalSorter(deps)
graph.prepare()
while graph.is_active():
ready = sorted(graph.get_ready())
print(ready)
graph.done(*ready)['extract_customers', 'extract_sales']
['load_customers', 'load_sales']
['build_gold']
['refresh_report']Four waves: both extracts, then both loads, then build_gold, then refresh_report. But it won't retry a failed task, backfill last month, alert anyone or keep run history.
That's Airflow's job. You define the workflow as a DAG in Python, and it runs tasks in dependency order, retries them, backfills past dates and shows every run in a web UI. The cost is running it, or a managed service.
dbt answers a different question: transformations as SQL select models in the warehouse, with dependencies, tests and documentation built in. If most logic is SQL on warehouse tables, dbt fits.
| Python scripts | Airflow | dbt | |
|---|---|---|---|
| WHAT IT IS | your code on a schedule | orchestrator: DAGs in Python | SQL models in the warehouse |
| DEPENDENCIES | you code them | task dependencies | between models |
| RETRIES, BACKFILLS | you code them | built in | dbt retry; a scheduler runs it |
| TESTS, HISTORY | you build them | every run in the UI | tests and docs built in |
Often it's all three: Python for extraction and APIs, dbt for transforms, Airflow to orchestrate.
Wrap the existing scripts as tasks first, then refactor.
Question 3: A Python job uses more memory every night until it crashes. How do you find and fix the leak?
First, confirm it's a leak. If each night is a fresh process, growth usually means growing input, like a full reload of a growing table. A real leak climbs run after run in one long-lived process, because something keeps references.
Here's a classic. A module-level dictionary caches every order and is never cleared. Module globals live as long as the process, so every night adds to it.
_seen: dict[int, dict] = {} # module level: lives forever
def run(night: int) -> int:
"""One nightly run inside a long-lived worker."""
for i in range(5_000):
order_id = night * 5_000 + i
_seen[order_id] = {"id": order_id, "status": "open"}
return len(_seen)To find it, we use tracemalloc: take a snapshot, run three nights, take another, and compare them by line.
import tracemalloc
from pathlib import Path
import job
tracemalloc.start()
before = tracemalloc.take_snapshot()
for night in range(1, 4): # three nights, one process
job.run(night)
after = tracemalloc.take_snapshot()
top = after.compare_to(before, "lineno")[0].traceback[0]
print(f"grows at {Path(top.filename).name}:{top.lineno}")grows at job.py:8The top growth is job.py:8, the line that fills the cache.
The fix: keep the cache local to the run, so it's freed when the run ends.
def run(night: int) -> int:
"""One run: the cache is freed when it returns."""
seen: dict[int, dict] = {}
for i in range(5_000):
order_id = night * 5_000 + i
seen[order_id] = {"id": order_id, "status": "open"}
return len(seen)Let's run the fixed job for three nights and measure it.
import job_fixed
start, _ = tracemalloc.get_traced_memory()
for night in range(1, 4):
job_fixed.run(night)
end, _ = tracemalloc.get_traced_memory()
tracemalloc.stop()
print("old job holds", len(job._seen), "orders")
print("fixed job grew over 1 MB:", end - start > 1_000_000)old job holds 15000 orders
fixed job grew over 1 MB: FalseThe old job still holds 15000 orders, and the fixed job didn't grow by more than a megabyte: False.
Other suspects to check:
- Caches with no max size.
- Logging handlers added every run.
- DataFrames appended to a global list.
Freed memory isn't always returned to the operating system, and a fresh process gives it all back.
Question 4: The business wants a new data source loaded by Friday. How do you scope and deliver it safely?
Before I promise Friday, I ask:
- What decision does it support?
- The must-have fields and the grain.
- The history, and how often it arrives.
- Access and a sample on day one, because access usually slips.
Day one, I profile the sample. Let's do that.
import pandas as pd
df = pd.read_csv("returns_sample.csv")
dt = pd.to_datetime(df["return_date"], format="ISO8601")
nulls = df.isna().sum()
print("rows:", len(df), "columns:", len(df.columns))
print("duplicate keys:", df.duplicated("return_id").sum())
print("nulls:", nulls[nulls > 0].to_dict())
print("dates:", dt.min().date(), "to", dt.max().date())rows: 8 columns: 5
duplicate keys: 1
nulls: {'reason': 2}
dates: 2026-10-01 to 2026-10-068 rows, 5 columns, 1 duplicate key, 2 missing reasons, and dates from 2026-10-01 to 2026-10-06. Those questions go to the source owner now, not on Thursday.
Then I cut the scope to a safe first slice: land the raw file as it is, load the must-have fields with an idempotent load, and add the checks. History and nice-to-have fields go in the next round, in writing.
| By Friday | Next round |
|---|---|
| raw file landed | full history |
| must-have fields, checks | nice-to-have fields |
Day by day:
- Monday: questions, access, a sample.
- Tuesday: profiling and the contract.
- Wednesday: build the load.
- Thursday: tests, and a reconcile in the test environment.
- Friday: the deploy, with monitoring and a short runbook.
A smaller, correct load on Friday beats a complete, wrong one.
Question 5: Your team's Python code has no tests and breaks often. What do you put in place first?
First, find out where it breaks. Read the last few incidents: bad input data, a code change, or an environment change. That tells you which guardrail pays off first. Then, in order:
- Pull requests and CI: ruff and pytest, starting from one smoke test, so the habit and the pipeline exist.
- A test for every bug fixed, and tests for the riskiest transformations, like money and dedup logic.
- Runtime data checks, to stop a bad batch before the load.
- A lock file, and a test environment before production.
Runtime checks matter because many failures are bad data, not bad code. This check collects every problem, and raises one clear error before anything is loaded.
class DataCheckError(Exception):
"""A batch broke a rule: stop before it is loaded."""
def check_batch(df: pd.DataFrame) -> None:
problems = []
if dups := df.duplicated("order_id").sum():
problems.append(f"{dups} duplicate order_id")
if missing := df["amount"].isna().sum():
problems.append(f"{missing} missing amount")
if problems:
raise DataCheckError("; ".join(problems))This step prints nothing; it prepares the data for the next step.
Let's run it on today's batch.
check_batch(pd.read_csv("orders_today.csv"))DataCheckError: 1 duplicate order_id; 2 missing amountDataCheckError: 1 duplicate order_id; 2 missing amount. Both problems in one message, and nothing loaded.
Within a few weeks, breaks turn into failed checks instead of wrong reports.
Don't chase a coverage number on day one. Start where the incidents are.
Question 6: A vendor changes their file format without notice and the load fails. How do you make the pipeline handle it?
Expect it to happen. Write down the contract: the columns and types you depend on. Then check every file against it before loading, and always land the raw file as it arrived, so you can reprocess it later.
The check reads a sample of the file and compares it with the contract: missing columns, new columns, and changed types.
EXPECTED = {"order_id": "int64", "order_date": "str",
"amount": "float64"}
def schema_drift(path):
got = pd.read_csv(path, nrows=1_000).dtypes.astype(str)
missing = sorted(EXPECTED.keys() - set(got.index))
new = sorted(set(got.index) - EXPECTED.keys())
changed = sorted(c for c, t in EXPECTED.items()
if c in got and got[c] != t)
return missing, new, changedThis step prints nothing; it prepares the data for the next step.
Let's check yesterday's file and today's.
files = ["orders_2026-10-22.csv", "orders_2026-10-23.csv"]
for name in files:
missing, new, changed = schema_drift(name)
print(name)
print(" missing:", missing)
print(" new:", new)
print(" type changed:", changed)orders_2026-10-22.csv
missing: []
new: []
type changed: []
orders_2026-10-23.csv
missing: ['amount']
new: ['channel', 'total_amount']
type changed: ['order_id']Yesterday's file matches: all three lists are empty. Today's file is missing amount, has two new columns, channel and total_amount, and order_id changed type.
Each kind of change gets a rule:
| Change | Rule |
|---|---|
| NEW COLUMN | load known columns, log, tell |
| MISSING OR RENAMED | stop, quarantine, alert |
| TYPE CHANGED | stop, quarantine, alert |
total_amount is probably amount renamed, but a person confirms that. Then you add a mapping in config, not a code change, and reprocess the quarantined file from the raw copy.
Agree a schema contract with the vendor, with notice before changes, and read columns by name, never by position.
Scenario 7: In the final round, the hiring manager asks you to design a Python ingestion framework for fifty sources. What is your design?
One framework, not fifty scripts. Every source follows the same path:
- Extract, with a connector per kind.
- Land the raw data as it arrived.
- Validate it against the contract.
- Load it into the warehouse, with a MERGE on the key.
- Write a load log, with the run ID and counts.
Each source is a config entry: connector kind, business key, load mode and options.
[sources.orders_api]
kind = "api"
key = "order_id"
mode = "incremental"
options = { endpoint = "/v1/orders" }
[sources.erp_customers]
kind = "db"
key = "customer_id"
mode = "full"
options = { table = "dbo.customers" }Connectors share one small interface, an extract method, and a registry, CONNECTORS, maps each kind to a class. Here, fakes stand in for the real API and database.
class Connector(Protocol):
def extract(self) -> list[dict]: ...
class FakeApi: # stands in for a REST API connector
def __init__(self, endpoint: str) -> None:
self.endpoint = endpoint
def extract(self) -> list[dict]:
return [{"order_id": n} for n in (1001, 1002, 1003)]One generic runner reads the config and runs every source the same way. Let's run it.
import tomllib
with open("sources.toml", "rb") as f:
sources = tomllib.load(f)["sources"]
for name, src in sources.items():
connector = CONNECTORS[src["kind"]](**src["options"])
rows = connector.extract()
key = src["key"]
bad = sum(row.get(key) is None for row in rows)
print(f"{name}: {len(rows)} rows, {bad} missing {key}")orders_api: 3 rows, 0 missing order_id
erp_customers: 4 rows, 1 missing customer_idorders_api gives 3 rows with 0 missing order_id. erp_customers gives 4 rows, 1 with no customer_id, so that batch goes to quarantine and the rest carry on.
Reliability is built in once, not per source:
- Idempotent loads that merge on the key.
- A load log with a run ID.
- Retries with backoff.
- Secrets from a vault.
An orchestrator like Airflow runs one task per source, with alerts and a dashboard of row counts and freshness.
Adding it is then a config entry and a test.
Key points to remember
- Loaded twice: communicate, find the load, fix in one transaction
- Scripts, Airflow or dbt: decide by the pain, migrate gradually
- Memory leak: tracemalloc, local state, a fresh process per run
- By Friday: a safe first slice; no tests: guardrails first
- Vendor changes: a contract; 50 sources: one config-driven framework
The full script
Here is everything in one script: the sample data and every step from this post, in order. Copy it and run it top to bottom in an empty folder, because it writes small sample files.
Show the full Python script (sample data and every step)
# ========================================================================
# Python Interview Q&A #10 - Real-world manager scenarios
# TechBrothersIT. Sample data is invented for teaching.
# Every step below was run with Python 3.14.6 and pandas 3.1.0; the outputs in the video and the blog
# post are the real outputs. Run it in an empty folder: it writes small sample files.
# ========================================================================
# Step 0: sample data
import sqlite3
from pathlib import Path
from typing import Protocol
# Q1: the warehouse table after yesterday's file was loaded twice
con = sqlite3.connect("warehouse.db", autocommit=False)
con.executescript("""
CREATE TABLE fact_sales (
sale_id INTEGER NOT NULL,
sale_date TEXT NOT NULL,
amount INTEGER NOT NULL,
source_file TEXT NOT NULL,
load_id INTEGER NOT NULL
);
-- load 41 and load 42 both loaded the Oct 22 file
INSERT INTO fact_sales VALUES
(5001, '2026-10-21', 120, 'sales_2026-10-21.csv', 40),
(5002, '2026-10-21', 80, 'sales_2026-10-21.csv', 40),
(5003, '2026-10-21', 100, 'sales_2026-10-21.csv', 40),
(5004, '2026-10-22', 150, 'sales_2026-10-22.csv', 41),
(5005, '2026-10-22', 50, 'sales_2026-10-22.csv', 41),
(5006, '2026-10-22', 40, 'sales_2026-10-22.csv', 41),
(5004, '2026-10-22', 150, 'sales_2026-10-22.csv', 42),
(5005, '2026-10-22', 50, 'sales_2026-10-22.csv', 42),
(5006, '2026-10-22', 40, 'sales_2026-10-22.csv', 42);
-- staging: today's file, as the job landed it
CREATE TABLE stg_sales (sale_id INTEGER, sale_date TEXT, amount INTEGER, source_file TEXT);
INSERT INTO stg_sales VALUES
(5007, '2026-10-23', 200, 'sales_2026-10-23.csv'),
(5008, '2026-10-23', 90, 'sales_2026-10-23.csv'),
(5009, '2026-10-23', 60, 'sales_2026-10-23.csv');
""")
con.commit()
UPSERT_SQL = """
INSERT INTO fact_sales
(sale_id, sale_date, amount, source_file, load_id)
SELECT sale_id, sale_date, amount, source_file, :load_id
FROM stg_sales
WHERE true -- SQLite: a WHERE is needed before ON CONFLICT
ON CONFLICT (sale_id) DO UPDATE SET
amount = excluded.amount,
load_id = excluded.load_id"""
# Q3: the leaky nightly job, and the fixed one
Path("job.py").write_text('''\
_seen: dict[int, dict] = {} # module level: lives forever
def run(night: int) -> int:
"""One nightly run inside a long-lived worker."""
for i in range(5_000):
order_id = night * 5_000 + i
_seen[order_id] = {"id": order_id, "status": "open"}
return len(_seen)
''', encoding="utf-8")
Path("job_fixed.py").write_text('''\
def run(night: int) -> int:
"""One run: the cache is freed when it returns."""
seen: dict[int, dict] = {}
for i in range(5_000):
order_id = night * 5_000 + i
seen[order_id] = {"id": order_id, "status": "open"}
return len(seen)
''', encoding="utf-8")
# Q4, Q5, Q6: sample files
Path("returns_sample.csv").write_text("""\
return_id,store_id,return_date,amount,reason
9001,12,2026-10-01,25.00,damaged
9002,12,2026-10-01,40.00,
9003,15,2026-10-02,15.50,wrong size
9003,15,2026-10-02,15.50,wrong size
9004,18,2026-10-03,60.00,
9005,12,2026-10-04,22.00,changed mind
9006,15,2026-10-05,35.00,damaged
9007,18,2026-10-06,18.00,wrong size
""", encoding="utf-8")
Path("orders_today.csv").write_text("""\
order_id,amount
1001,120.00
1002,
1003,80.00
1003,80.00
1004,
1005,45.00
""", encoding="utf-8")
Path("orders_2026-10-22.csv").write_text("""\
order_id,order_date,amount
1001,2026-10-22,120.00
1002,2026-10-22,75.50
""", encoding="utf-8")
Path("orders_2026-10-23.csv").write_text("""\
order_id,order_date,channel,total_amount
ORD-1003,2026-10-23,web,64.00
ORD-1004,2026-10-23,store,210.00
""", encoding="utf-8")
# Q7: the source config, and connectors (fakes stand in for the real API and database)
Path("sources.toml").write_text('''\
[sources.orders_api]
kind = "api"
key = "order_id"
mode = "incremental"
options = { endpoint = "/v1/orders" }
[sources.erp_customers]
kind = "db"
key = "customer_id"
mode = "full"
options = { table = "dbo.customers" }
''', encoding="utf-8")
class Connector(Protocol):
def extract(self) -> list[dict]: ...
class FakeApi: # stands in for a REST API connector
def __init__(self, endpoint: str) -> None:
self.endpoint = endpoint
def extract(self) -> list[dict]:
return [{"order_id": n} for n in (1001, 1002, 1003)]
class FakeDb: # stands in for a database connector
def __init__(self, table: str) -> None:
self.table = table
def extract(self) -> list[dict]:
return [{"customer_id": n} for n in (21, None, 23, 24)]
CONNECTORS: dict[str, type[Connector]] = {"api": FakeApi, "db": FakeDb}
# Step 1: Find the duplicate load: rows and totals per file and load
BY_LOAD = """
SELECT source_file, load_id, COUNT(*), SUM(amount)
FROM fact_sales
GROUP BY source_file, load_id
ORDER BY load_id"""
for file, load_id, rows, total in con.execute(BY_LOAD):
print(f"{file} load {load_id}: {rows} rows, {total}")
# Step 2: Fix it in one transaction: delete the extra load, add a unique key
with con: # one transaction: commit, or roll back on error
cur = con.execute(
"DELETE FROM fact_sales WHERE load_id = ?", (42,))
con.execute("CREATE UNIQUE INDEX ux_fact_sales_sale_id"
" ON fact_sales (sale_id)")
print("deleted rows:", cur.rowcount)
for file, load_id, rows, total in con.execute(BY_LOAD):
print(f"{file} load {load_id}: {rows} rows, {total}")
# Step 3: Prove the new load is safe to rerun: the same file twice
DAY_TOTAL = """
SELECT COUNT(*), SUM(amount) FROM fact_sales
WHERE sale_date = ?"""
day = "2026-10-23"
for load_id in (43, 44): # the same file, loaded twice
with con:
con.execute(UPSERT_SQL, {"load_id": load_id})
n, total = con.execute(DAY_TOTAL, (day,)).fetchone()
print(f"after load {load_id}: {n} rows, total {total}")
con.close() # with con: commits or rolls back, never closes
# Step 4: What an orchestrator does first: run tasks in dependency order
from graphlib import TopologicalSorter
deps = {"load_sales": ["extract_sales"],
"load_customers": ["extract_customers"],
"build_gold": ["load_sales", "load_customers"],
"refresh_report": ["build_gold"]}
graph = TopologicalSorter(deps)
graph.prepare()
while graph.is_active():
ready = sorted(graph.get_ready())
print(ready)
graph.done(*ready)
# Step 5: Find the line whose memory keeps growing with tracemalloc
import tracemalloc
from pathlib import Path
import job
tracemalloc.start()
before = tracemalloc.take_snapshot()
for night in range(1, 4): # three nights, one process
job.run(night)
after = tracemalloc.take_snapshot()
top = after.compare_to(before, "lineno")[0].traceback[0]
print(f"grows at {Path(top.filename).name}:{top.lineno}")
# Step 6: The fixed job does not grow
import job_fixed
start, _ = tracemalloc.get_traced_memory()
for night in range(1, 4):
job_fixed.run(night)
end, _ = tracemalloc.get_traced_memory()
tracemalloc.stop()
print("old job holds", len(job._seen), "orders")
print("fixed job grew over 1 MB:", end - start > 1_000_000)
# Step 7: Day one: profile the sample file of the new source
import pandas as pd
df = pd.read_csv("returns_sample.csv")
dt = pd.to_datetime(df["return_date"], format="ISO8601")
nulls = df.isna().sum()
print("rows:", len(df), "columns:", len(df.columns))
print("duplicate keys:", df.duplicated("return_id").sum())
print("nulls:", nulls[nulls > 0].to_dict())
print("dates:", dt.min().date(), "to", dt.max().date())
# Step 8: A runtime data check that stops a bad batch
class DataCheckError(Exception):
"""A batch broke a rule: stop before it is loaded."""
def check_batch(df: pd.DataFrame) -> None:
problems = []
if dups := df.duplicated("order_id").sum():
problems.append(f"{dups} duplicate order_id")
if missing := df["amount"].isna().sum():
problems.append(f"{missing} missing amount")
if problems:
raise DataCheckError("; ".join(problems))
# Step 9: Run the check on today's batch
# This step is EXPECTED TO FAIL: it shows the error the video explains.
try:
check_batch(pd.read_csv("orders_today.csv"))
except Exception as e:
print(type(e).__name__ + ":", e)
# Step 10: Schema drift check against the contract
EXPECTED = {"order_id": "int64", "order_date": "str",
"amount": "float64"}
def schema_drift(path):
got = pd.read_csv(path, nrows=1_000).dtypes.astype(str)
missing = sorted(EXPECTED.keys() - set(got.index))
new = sorted(set(got.index) - EXPECTED.keys())
changed = sorted(c for c, t in EXPECTED.items()
if c in got and got[c] != t)
return missing, new, changed
# Step 11: Check yesterday's and today's vendor files
files = ["orders_2026-10-22.csv", "orders_2026-10-23.csv"]
for name in files:
missing, new, changed = schema_drift(name)
print(name)
print(" missing:", missing)
print(" new:", new)
print(" type changed:", changed)
# Step 12: One config-driven runner for every source
import tomllib
with open("sources.toml", "rb") as f:
sources = tomllib.load(f)["sources"]
for name, src in sources.items():
connector = CONNECTORS[src["kind"]](**src["options"])
rows = connector.extract()
key = src["key"]
bad = sum(row.get(key) is None for row in rows)
print(f"{name}: {len(rows)} rows, {bad} missing {key}")
All 10 parts are listed on the Real Time Python for Data Engineers Interview Questions and Answers page on this blog.
The sample data on this page is invented for teaching. Every output is the real output of the code above it, run with Python 3.14.6 and pandas 3.1.0. Part of the Real Time Python for Data Engineers Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Python and library documentation for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.