SESSION 8

Python Internals · FastAPI Deep Dive · System Design for ML · DSA for ML Roles

Ishaan Goyal — Interview Preparation

■ Concept ■ Say This ■ When to Use ■ Traps ■ Code ■ Key Question

SECTION A — PYTHON INTERNALS

■ KEY What is Python's GIL? How does it affect your ML code?

■ CONCEPT — What the interviewer is testing

What the GIL is:

The Global Interpreter Lock is a mutex that protects access to Python objects — only ONE thread can execute
Python bytecode at a time in a single process. CPython (the standard Python) uses reference counting for
memory management, and the GIL protects those reference counts from race conditions.

What the GIL means in practice:

• Threading in Python does NOT give CPU parallelism for CPU-bound work

• Two threads trying to run Python code simultaneously: one runs, one waits for the GIL

• Threading DOES help for I/O-bound work: while one thread waits for I/O (network, disk), it releases the GIL
— other threads run

• This is why async/await (single thread, event loop) is preferred over threading for I/O-bound FastAPI

How to bypass the GIL for CPU-bound ML work:

• multiprocessing: separate processes, each with own GIL — true CPU parallelism (Gunicorn workers)

• NumPy/PyTorch: C extensions release the GIL during computation — matrix operations run in parallel

• Cython with nogil: write performance-critical sections in Cython, release GIL explicitly

• [Link]: process pool for parallel CPU work

The GIL and your FastAPI deployment:

• Single Uvicorn process: GIL limits CPU parallelism — but async I/O still works well

• Gunicorn + multiple workers: bypasses GIL — each worker is a separate process, own GIL

• PyTorch model inference: releases GIL during C++ computation — can overlap with Python

■ SAY THIS — Word for word

"The GIL is a mutex in CPython that allows only one thread to execute Python bytecode at a time. For
CPU-bound work like model inference in pure Python, threading gives no parallelism — threads take turns. For
I/O-bound work like database calls or API requests, threads work fine because waiting for I/O releases the GIL.
In my deployment I bypass the GIL using Gunicorn — each worker is a separate OS process with its own GIL,
giving true CPU parallelism. PyTorch's model inference itself releases the GIL during C++ computation, so
Python code can run concurrently with the GPU/CPU computation. That's why run_in_executor works for
offloading inference."

■ FOLLOW-UP TRAPS + ANSWERS

Q: Threads vs Processes vs Async — when to use each in Python?

• Async (asyncio): many I/O-bound tasks, single thread, no GIL issue, lowest overhead. Best for FastAPI, web
scrapers, API clients.

• Threading: I/O-bound work with blocking libraries that don't support async (old DB drivers, legacy code). GIL
released during I/O.

• Multiprocessing: CPU-bound work needing true parallelism. Separate memory space — no shared state.
Use for: parallel data preprocessing, model training on multiple datasets.

• Rule: async first, threading if library doesn't support async, multiprocessing for CPU-heavy.

Q: Python memory management — how does garbage collection work?

CPython uses reference counting as primary mechanism: every object tracks how many references point to it.
When count reaches 0, object is immediately deallocated. Problem: circular references (A → B → A) never
reach 0. Solution: cyclic garbage collector (gc module) runs periodically to detect and break cycles. del keyword
decrements reference count — if 0, immediately frees memory. Unlike Java GC, Python's reference counting is
deterministic.

Q: What is a Python generator and when would you use one for ML?

A generator is a function that yields values one at a time instead of returning all at once. Uses lazy evaluation —
values computed on demand. Key for ML: data loading. Instead of loading 500K records into memory at once, a
generator yields one batch at a time. PyTorch DataLoader uses generators internally. Also: processing large log
files line by line without loading entire file.
Q2 What are decorators in Python? Write one from scratch.

■ CONCEPT — What the interviewer is testing

What a decorator is:

A decorator is a function that takes another function as input, wraps it with additional behaviour, and returns the
modified function. Syntactic sugar for: func = decorator(func). Uses: logging, timing, caching, authentication,
retry logic, rate limiting.

Why they matter for ML engineering:

• @[Link]('/predict'): FastAPI uses decorators to register route handlers

• @torch.no_grad(): disables gradient tracking during inference — saves memory and computation

• @lru_cache: memoises expensive function calls (embedding computation, model loading)

• Custom: @retry, @timer, @validate_input — production ML API patterns

■ CODE — Decorators from scratch — 4 patterns you must know

import time, functools

from typing import Callable

# ■■ PATTERN 1: Basic decorator ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■

def timer(func):

@[Link](func) # preserves func name and docstring

def wrapper(*args, **kwargs):

start = [Link]()

result = func(*args, **kwargs)

print(f'{func.__name__} took {[Link]()-start:.3f}s')

return result

return wrapper

@timer

def run_inference(text): return [Link](text)

# Same as: run_inference = timer(run_inference)

# ■■ PATTERN 2: Decorator with arguments ■■■■■■■■■■■■■■■■■■■■■■■

def retry(max_attempts=3, delay=1.0):

def decorator(func):

@[Link](func)

def wrapper(*args, **kwargs):

for attempt in range(max_attempts):

try:

return func(*args, **kwargs)

except Exception as e:

if attempt == max_attempts - 1: raise

[Link](delay)

return wrapper

return decorator

@retry(max_attempts=3, delay=0.5)

def call_openai_api(prompt): ...

# ■■ PATTERN 3: Class-based decorator ■■■■■■■■■■■■■■■■■■■■■■■■■■

class RateLimiter:

def __init__(self, calls_per_sec):

self.min_interval = 1.0 / calls_per_sec

self.last_call = 0

def __call__(self, func):

@[Link](func)

def wrapper(*args, **kwargs):

elapsed = [Link]() - self.last_call

if elapsed < self.min_interval:

[Link](self.min_interval - elapsed)

self.last_call = [Link]()
 return func(*args, **kwargs)

return wrapper

# ■■ PATTERN 4: @lru_cache — built-in memoisation ■■■■■■■■■■■■■■

from functools import lru_cache

@lru_cache(maxsize=1000)

def get_embedding(text: str) -> tuple:

return tuple([Link](text)) # cached — same text = no API call

■ SAY THIS — Word for word

"A decorator is a higher-order function — it takes a function, wraps it with additional behaviour, and returns the
wrapped version. The @ syntax is shorthand for func = decorator(func). In my ML projects I use decorators
constantly: FastAPI's @[Link]('/predict') registers my route handler, @torch.no_grad() disables gradient
tracking during inference saving memory, and I wrote a custom @retry decorator for OpenAI API calls that
retries up to 3 times with exponential backoff on rate limit errors. The key implementation detail is
[Link] — without it the wrapper function loses the original function's name and docstring, which breaks
debugging and FastAPI's auto-documentation."

■ FOLLOW-UP TRAPS + ANSWERS

Q: @staticmethod vs @classmethod vs regular method?

• Regular method: receives self (instance). Accesses instance state.

• @classmethod: receives cls (class itself). Can create alternative constructors. Model.from_config(config).

• @staticmethod: receives neither. Pure utility function — logically belongs to class but doesn't need instance
or class. Preprocessing helper methods.

Q: What is [Link] and why is it critical?

Without @[Link](func): the wrapper function replaces the original — its __name__ is 'wrapper', its
__doc__ is None, its __module__ is the decorator's module. FastAPI reads __name__ to generate route IDs and
OpenAPI docs. If two routes have the same name ('wrapper'), FastAPI raises an error. [Link] copies all
metadata from the wrapped function to the wrapper.

Q: What is a context manager? Write one.

Context managers define setup and teardown for a with block. __enter__ runs at start, __exit__ runs at end
(even on exception). Used for: file handling, DB connections, GPU memory management, timing. ML use:
torch.no_grad() is a context manager — disables grad on enter, re-enables on exit. Write with @contextmanager
decorator from contextlib for simple cases.
SECTION B — GENERATORS, COMPREHENSIONS & DATA
STRUCTURES

What is a generator? How does it differ from a list? Write a data pipeline using
Q3
generators.

■ CONCEPT — What the interviewer is testing

List vs Generator — memory:

• List: [x*2 for x in range(1_000_000)] — creates 1M integers in memory instantly. ~8MB.

• Generator: (x*2 for x in range(1_000_000)) — creates a lazy iterator. ~200 bytes.

• Generator computes each value only when next() is called — lazy evaluation.

• Cannot index a generator (gen[5] fails). Can only iterate once — not reusable.

yield keyword — how generators work:

• A function with yield is a generator function. Calling it returns a generator object.

• Each call to next() runs until the next yield — then pauses, saving local state.

• StopIteration raised when function returns or runs out of yields.

ML use cases for generators:

• Data loading: yield one batch at a time from disk — never load full dataset into RAM

• Streaming predictions: yield prediction per row as it's computed

• Text preprocessing pipeline: chain generators for each step — memory-efficient

• PyTorch DataLoader: uses __iter__ and __len__ protocol — same concept

■ CODE — Generator-based ML data pipeline

import pandas as pd

import numpy as np

from pathlib import Path

# ■■ Generator 1: load CSV in chunks (never loads full file) ■■■

def load_chunks(filepath: str, chunksize: int = 1000):

for chunk in pd.read_csv(filepath, chunksize=chunksize):

 yield chunk # yields one DataFrame at a time

# ■■ Generator 2: preprocess each chunk ■■■■■■■■■■■■■■■■■■■■■■■■

def preprocess(chunks):

for chunk in chunks:

chunk = [Link]()

chunk['age'] = 2024 - chunk['year']

yield chunk

# ■■ Generator 3: extract features ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■

def extract_features(chunks):

for chunk in chunks:

X = chunk[['age', 'km_driven', 'engine']].values

y = chunk['price'].values

yield X, y

# ■■ Pipeline: chain generators — each step lazy ■■■■■■■■■■■■■■■■

def train_pipeline(filepath):

chunks = load_chunks(filepath, chunksize=1000)

cleaned = preprocess(chunks)

batches = extract_features(cleaned)

for X_batch, y_batch in batches: # pulls one batch at a time

model.partial_fit(X_batch, y_batch) # online learning

# ■■ yield from — delegate to sub-generator ■■■■■■■■■■■■■■■■■■■■

def load_all_files(directory: str):

for filepath in Path(directory).glob('*.csv'):

yield from load_chunks(str(filepath)) # flatten nested generator

# ■■ Generator expression in pipeline ■■■■■■■■■■■■■■■■■■■■■■■■■■

files = Path('data/').glob('*.csv') # generator

lines = (line for f in files # nested generator expression

for line in open(f))

processed = ([Link]() for line in lines if [Link]())

■ SAY THIS — Word for word

"A generator is a lazy iterator — instead of computing all values upfront like a list, it computes each value on
demand when next() is called. A list of 1 million items takes 8MB; the equivalent generator takes 200 bytes. For
ML, this is critical: I use generators to load training data in batches from disk rather than loading 500K rows into
RAM at once. I chain generators for each preprocessing step — load chunk, clean, extract features — and each
step processes one batch at a time. The entire pipeline uses constant memory regardless of dataset size.
PyTorch's DataLoader uses this same pattern internally."

■ FOLLOW-UP TRAPS + ANSWERS

Q: List comprehension vs generator expression vs map — when to use each?

• List comprehension [x for x in ...]: when you need the full list, will access multiple times, need len(), indexing.

• Generator expression (x for x in ...): when iterating once, large data, memory matters, chaining with other
iterators.

• map(func, iterable): functional style, lazy like generator, slightly faster than generator expression for simple
functions, less readable.

• Rule: generator expression for data pipelines, list comprehension when you need the whole result.

Q: What is itertools and name 3 useful functions for ML?

• [Link](gen, n): take first n items from generator — for testing pipeline with subset

• [Link](*iterables): flatten multiple generators into one — merge multiple CSV generators

• [Link](iterable, n) [Python 3.12]: group items into batches of size n — easier than manual
chunking

• [Link](a, b): Cartesian product — grid search hyperparameter combinations without nested loops
SECTION C — FASTAPI INTERNALS & PRODUCTION PATTERNS

■ KEY How does FastAPI work internally? What makes it fast?

■ CONCEPT — What the interviewer is testing

FastAPI's speed — 4 reasons:

• 1. ASGI (Async Server Gateway Interface): async-native framework. Requests handled as coroutines —
event loop juggles thousands of concurrent I/O-bound requests without threads.

• 2. Pydantic v2: data validation compiled to Rust — 5-50x faster than pure Python validation. Every request
body is validated and typed at near-C speed.

• 3. Starlette underneath: FastAPI is built on Starlette — a lightweight, high-performance ASGI framework.

FastAPI adds dependency injection and OpenAPI on top.

• 4. No overhead: direct path to your function — no middleware layers, no ORM by default.

Dependency Injection — the most powerful FastAPI feature:

Dependencies are functions that run before your route handler and inject shared resources. FastAPI resolves
them automatically — call the same dependency from 100 routes, it runs once per request (or once per app with
Depends(func, use_cache=True)).

Lifespan events — correct way to load your model:

• @asynccontextmanager on lifespan function: code before yield runs at startup, after yield runs at shutdown

• Load model ONCE at startup — store in [Link]

• Never load model per request — would take seconds and defeat the purpose of serving

■ CODE — Production FastAPI — model loading, dependencies, middleware

from fastapi import FastAPI, Depends, HTTPException, Request

from [Link] import CORSMiddleware

from [Link] import TrustedHostMiddleware

from pydantic import BaseModel, Field, validator

from contextlib import asynccontextmanager

import time, logging

import [Link]
# ■■ Lifespan: load model once at startup ■■■■■■■■■■■■■■■■■■■■■■

@asynccontextmanager

async def lifespan(app: FastAPI):

# STARTUP

[Link] = [Link].load_model('models:/CarPrice/Production')

[Link] = load_scaler('[Link]')

[Link]('Model loaded successfully')

yield

# SHUTDOWN

[Link]('Shutting down')

app = FastAPI(title='Car Price API', version='1.0', lifespan=lifespan)

# ■■ Middleware: CORS, logging, timing ■■■■■■■■■■■■■■■■■■■■■■■■■

app.add_middleware(CORSMiddleware,

allow_origins=['[Link]

allow_methods=['POST', 'GET'], allow_headers=['*'])

@[Link]('http')

async def log_requests(request: Request, call_next):

start = [Link]()

response = await call_next(request)

duration = [Link]() - start

[Link](f'{[Link]} {[Link]} {response.status_code} {duration:.3f}s')

return response

# ■■ Pydantic models — strict validation ■■■■■■■■■■■■■■■■■■■■■■■

class CarFeatures(BaseModel):

brand: str = Field(..., min_length=1, max_length=50)

year: int = Field(..., ge=1990, le=2024)

 km_driven: float = Field(..., ge=0, le=1_000_000)

fuel: str = Field(..., pattern='^(Petrol|Diesel|Electric|CNG)$')

@validator('year')

def year_not_future(cls, v):

if v > 2024: raise ValueError('Year cannot be in the future')

return v

class Prediction(BaseModel):

price_lakhs: float

confidence: str

model_version: str

# ■■ Dependency: get model from app state ■■■■■■■■■■■■■■■■■■■■■■

def get_model(request: Request):

return [Link]

def get_scaler(request: Request):

return [Link]

# ■■ Route with dependencies ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■

@[Link]('/predict', response_model=Prediction)

async def predict(

features: CarFeatures,

model=Depends(get_model),

scaler=Depends(get_scaler)

):

import asyncio

from [Link] import ThreadPoolExecutor

executor = ThreadPoolExecutor(max_workers=4)
 def _inference():

X = engineer_features(features)

X_scaled = [Link](X)

return float([Link](X_scaled)[0])

loop = asyncio.get_event_loop()

price = await loop.run_in_executor(executor, _inference)

return Prediction(price_lakhs=round(price, 2),

confidence='high' if price > 3 else 'medium',

model_version='1.3')

# ■■ Health check for load balancer ■■■■■■■■■■■■■■■■■■■■■■■■■■■■

@[Link]('/health')

async def health(request: Request):

model_loaded = hasattr([Link], 'model')

return {'status': 'ok' if model_loaded else 'degraded', 'model': model_loaded}

■ SAY THIS — Word for word

"FastAPI is fast for three reasons. First, it's built on ASGI and Starlette — async-native, so the event loop
handles thousands of concurrent I/O-bound requests without threads. Second, Pydantic v2 compiles data
validation to Rust — every incoming request body is validated and typed at near-C speed. Third, it uses
dependency injection to manage shared resources. In production I load the model once at startup using the
lifespan context manager and store it in [Link] — never per request. Dependencies inject the model into
route handlers without coupling them to global state. I also add request logging middleware to track every
endpoint's latency in CloudWatch."

■ FOLLOW-UP TRAPS + ANSWERS

Q: What is Pydantic and what problem does it solve?

Pydantic is a data validation library using Python type annotations. Problem: JSON from clients can contain
wrong types, missing fields, invalid values. Without Pydantic: you manually check 'if year not in data: raise error'.
With Pydantic: define a class with typed fields, FastAPI automatically validates incoming JSON against it and
returns a detailed 422 error if invalid. Also handles serialisation — Python objects to JSON. Pydantic v2 (Rust
core) is 5-50x faster than v1.

Q: FastAPI dependency injection — what is it and why use it?

Dependencies are callables (functions, classes) that FastAPI calls before your route handler and injects the
return value. Benefits: (1) Shared resources (model, DB connection, auth check) defined once, used in many
routes. (2) Testability — override dependencies in tests (swap real model for mock). (3) Automatic cleanup —
yield-based dependencies run teardown after request completes. Example: DB session opened in dependency,
yielded to handler, closed after response.

Q: How do you add authentication to your FastAPI ML API?

API key auth (simple): check X-API-Key header in a dependency, raise 403 if invalid. JWT auth: decode and
verify JWT token in dependency, extract user claims. OAuth2: FastAPI has built-in OAuth2PasswordBearer. For
ML APIs: API key is most common — generate keys per client, store hashed in DB, verify in dependency
injected into protected routes.

Q: What is middleware in FastAPI and how does it differ from a dependency?

Middleware runs for EVERY request/response, regardless of route — good for logging, timing, CORS,
compression, auth headers. Dependencies run for SPECIFIC routes they're injected into — good for route-level
auth, model loading, DB sessions. Middleware sees raw request/response objects. Dependencies have access
to typed, validated data. Use both: middleware for cross-cutting concerns, dependencies for route-specific logic.
SECTION D — ML SYSTEM DESIGN

■ KEY Design an ML inference system that handles 10,000 requests/day reliably.

■ CONCEPT — What the interviewer is testing

Requirements clarification first — always ask:

• Latency requirement: <100ms? <1s? <10s? (determines sync vs async serving)

• Throughput: 10K/day = ~7 req/min average, but what is peak? (10x = 70 req/min?)

• Model size: 100MB sklearn or 10GB LLM? (determines instance type)

• Availability: 99.9% (8.7h downtime/year) or 99.99% (52min/year)?

• Data privacy: can data leave the server? (determines logging policy)

Architecture — 5 components:

1. API Layer (FastAPI + Gunicorn + Nginx):

• FastAPI with Pydantic validation and async request handling

• Gunicorn: 2×cores+1 Uvicorn workers per EC2

• Nginx: reverse proxy, SSL termination, rate limiting (10 req/min per IP)

2. Model Serving:

• Load model once at startup from S3 or MLflow Model Registry

• CPU inference: run_in_executor offloads to thread pool

• GPU inference: Triton Inference Server or TorchServe for high throughput

• Request batching: collect requests for 50ms, run as batch — 3-5x throughput improvement

3. Caching Layer (Redis):

• Cache predictions: hash(input features) → prediction. 30% hit rate = 30% less compute.

• Semantic cache for LLM: similar queries return cached response

• Rate limit counters: per-user request tracking

4. Async Task Queue (Celery + Redis) — for long inference:

• Receive request → return job_id immediately (202 Accepted)

• Celery worker processes inference in background

• Client polls GET /result/{job_id} or receives WebSocket push

• Use for: inference > 2 seconds, batch jobs, report generation

5. Observability (CloudWatch + structured logging):

• Log every request: input hash, latency, model version, prediction

• CloudWatch alarms: p99 latency > 2s, error rate > 1%, CPU > 70%

• Prediction distribution monitoring: detect model drift

■ CODE — System design — request flow

# Request flow for sync inference (< 2 seconds):

Client --> Nginx (rate limit, SSL) --> FastAPI (validate) --> Redis cache?

Cache HIT: return cached

Cache MISS:

ThreadPoolExecutor ([Link])

Store in Redis cache (TTL=1hr)

Return prediction + log to CloudWatch

# Request flow for async inference (> 2 seconds):

Client --> FastAPI --> return {job_id: 'abc123'} immediately (202)

--> publish task to Redis queue

Celery worker picks up task

--> run_model() --> store result in Redis

Client polls GET /result/abc123

--> FastAPI reads result from Redis --> return 200

# Auto Scaling trigger:

CloudWatch: CPU > 70% for 2min --> ASG alarm --> launch new EC2

New EC2: UserData script --> pull Docker image from ECR

--> docker-compose up --> register with ALB --> serve traffic

Total time: ~3 minutes from alarm to serving

■ SAY THIS — Word for word

"I'd design this in five layers. Client requests hit Nginx first for SSL termination and rate limiting. FastAPI
validates the request with Pydantic — invalid inputs return 422 immediately without touching the model. Next,
Redis cache check: hash the input features, if we've seen this exact input before return the cached prediction
instantly. Cache miss goes to the model: for fast inference I offload to a thread pool with run_in_executor, for
slow inference I return a job ID and process via Celery. All predictions are logged with latency and input hash to
CloudWatch for monitoring. The infrastructure runs on EC2 behind an ALB with Auto Scaling — CloudWatch
CPU alarm at 70% adds instances automatically. Minimum 2 instances across two availability zones for high
availability."

■ FOLLOW-UP TRAPS + ANSWERS

Q: How do you do A/B testing for ML models?

Traffic splitting: ALB routes X% to model A server, (100-X)% to model B server. Or: within one server, use
feature flags — hash user_id % 100, if < 20 use model B. Log which model version served each request.
Compare: latency, error rate, downstream business metric (conversion rate, click-through). Statistical
significance test (t-test, Mann-Whitney) before declaring a winner. Gradual rollout: 5% → 20% → 50% → 100%.

Q: How do you handle model versioning in production?

MLflow Model Registry: Staging → Production → Archived. Serving code always loads 'Production' stage — no
hardcoded model paths. To deploy new model: register in Registry, test in Staging (shadow mode — run
alongside production, compare outputs), promote to Production via Registry UI or API. Rollback: demote new
model, promote old model. Zero code changes needed.

Q: What is shadow mode deployment?

Run new model in parallel with production model — both receive the same requests, but only the production
model's response is returned to the user. New model's predictions are logged and compared offline. Lets you
validate new model on real production traffic with zero risk. After confirming new model performs better, promote
it to production.
Q: How do you handle a model that takes 30 seconds to run inference?

Async task queue is mandatory — cannot hold HTTP connection open for 30 seconds (timeout). Pattern: POST
/predict → return {job_id} immediately with 202 Accepted. Celery worker processes inference. Client either polls
GET /result/{job_id} every 5 seconds, or gets a WebSocket push when complete. Store result in Redis with
TTL=1hr. This pattern scales to arbitrary inference time.
SECTION E — DSA PATTERNS FOR ML ENGINEERING INTERVIEWS

What DSA topics come up in ML engineering interviews? What patterns must you
Q6
know?

■ CONCEPT — What the interviewer is testing

Reality of DSA in ML engineering interviews:

ML engineering roles ask lighter DSA than pure SWE roles — but you WILL get coding questions. Focus:
efficient data processing, tree traversal (decision trees), graph problems (dependency graphs), sliding window
(time series), hash maps (feature lookups). Complexity analysis matters — O(n log n) vs O(n²) for 500K records.

5 most common patterns for ML roles:

• 1. Hash Map / Dict: O(1) lookup for feature encoding, count occurrences, group by

• 2. Sliding Window: time series feature extraction, moving averages, rolling statistics

• 3. Two Pointers: merging sorted arrays, finding pairs, partition data

• 4. Binary Search: find optimal hyperparameter in sorted search space, find percentile

• 5. Tree/Graph traversal: model architecture graphs, dependency resolution, DAG processing

Complexity you must know for ML context:

• pandas groupby: O(n log n) — sorting + aggregation

• dict lookup: O(1) average, O(n) worst case (hash collision)

• sklearn KMeans: O(n * k * d * iterations) where n=samples, k=clusters, d=dimensions

• Nearest neighbour (brute force): O(n * d). HNSW: O(d * log n)

• Sorting 500K records: O(n log n) ≈ 500K * 19 ≈ 9.5M operations — fast

■ CODE — 5 DSA patterns every ML engineer must code

# ■■ PATTERN 1: Frequency count with hash map ■■■■■■■■■■■■■■■■■■

# Count category distribution in dataset — O(n)

from collections import Counter, defaultdict

brand_counts = Counter(df['brand']) # {BMW: 500, Maruti: 2000, ...}

# Or: group and aggregate

brand_stats = defaultdict(list)

for _, row in [Link]():

brand_stats[row['brand']].append(row['price'])

brand_mean = {k: sum(v)/len(v) for k, v in brand_stats.items()}

# ■■ PATTERN 2: Sliding window — rolling average ■■■■■■■■■■■■■■■■

# Rolling 7-day average prediction (time series) — O(n)

def rolling_average(predictions: list, window: int) -> list:

result = []

window_sum = sum(predictions[:window])

[Link](window_sum / window)

for i in range(window, len(predictions)):

window_sum += predictions[i] - predictions[i - window]

[Link](window_sum / window)

return result

# O(n) — single pass. Naive: O(n*w).

# ■■ PATTERN 3: Binary search — find percentile threshold ■■■■■■■

# Find min cluster count where silhouette > threshold — O(log n)

import bisect

def find_optimal_k(silhouette_scores: list, threshold: float) -> int:

lo, hi = 2, len(silhouette_scores) + 2

while lo < hi:

mid = (lo + hi) // 2

if compute_silhouette(mid) >= threshold:

hi = mid

else:

lo = mid + 1

return lo
# ■■ PATTERN 4: Two pointers — merge sorted feature lists ■■■■■■■

def merge_sorted_features(a: list, b: list) -> list:

result, i, j = [], 0, 0

while i < len(a) and j < len(b):

if a[i] <= b[j]: [Link](a[i]); i += 1

else: [Link](b[j]); j += 1

return result + a[i:] + b[j:] # O(n+m)

# ■■ PATTERN 5: BFS/DFS — topological sort for pipeline stages ■■

from collections import deque

def topological_sort(stages: dict) -> list:

# stages = {'train': ['preprocess'], 'evaluate': ['train']}

in_degree = {s: 0 for s in stages}

for s, deps in [Link]():

for dep in deps: in_degree[s] += 1

queue = deque([s for s, d in in_degree.items() if d == 0])

order = []

while queue:

node = [Link](); [Link](node)

for s, deps in [Link]():

if node in deps:

in_degree[s] -= 1

if in_degree[s] == 0: [Link](s)

return order

■ SAY THIS — Word for word

"ML engineering interviews focus on practical DSA — efficient data processing, not trick LeetCode hard
problems. The five patterns I focus on: hash maps for O(1) feature lookups and frequency counting; sliding
window for time series rolling statistics in O(n) instead of O(n×w); binary search for finding optimal thresholds or
hyperparameters in sorted search spaces; two pointers for merging sorted data efficiently; and BFS/topological
sort for processing ML pipeline dependency graphs — which is literally what DVC uses internally to decide which
stages to rerun. I always think about complexity for ML scale: an O(n²) algorithm on 500K records is 250 billion
operations — unacceptable. The same task in O(n log n) is 9.5 million."

Q7 Given a list of predictions and actual values, compute RMSE without using sklearn.

■ CONCEPT — What the interviewer is testing

Why this is asked:

Tests: (1) Do you understand what RMSE actually computes, not just call a function? (2) Can you write clean,
efficient Python? (3) Do you handle edge cases (empty list, division by zero)? (4) Numpy vectorised thinking.

RMSE formula: sqrt(mean((predictions - actuals)²))

• Step 1: compute element-wise squared differences

• Step 2: compute mean of those squared differences

• Step 3: take square root

• Edge cases: empty lists, mismatched lengths, NaN values

■ CODE — RMSE — 4 implementations from naive to production

import numpy as np

import math

# ■■ Version 1: Pure Python (show you understand the formula) ■■

def rmse_pure(actuals: list, predictions: list) -> float:

if len(actuals) != len(predictions):

raise ValueError(f'Length mismatch: {len(actuals)} vs {len(predictions)}')

if len(actuals) == 0:

raise ValueError('Empty lists')

squared_errors = [(a - p) ** 2 for a, p in zip(actuals, predictions)]

mse = sum(squared_errors) / len(squared_errors)

 return [Link](mse)

# ■■ Version 2: NumPy vectorised (what you'd use in practice) ■■

def rmse_numpy(actuals, predictions) -> float:

a = [Link](actuals, dtype=np.float64)

p = [Link](predictions, dtype=np.float64)

if [Link] != [Link]:

raise ValueError('Shape mismatch')

return float([Link]([Link]((a - p) ** 2)))

# ■■ Version 3: Handle NaN (production robustness) ■■■■■■■■■■■■■

def rmse_robust(actuals, predictions) -> float:

a, p = [Link](actuals, float), [Link](predictions, float)

mask = ~([Link](a) | [Link](p)) # ignore NaN pairs

if [Link]() == 0: raise ValueError('All values are NaN')

return float([Link]([Link]((a[mask] - p[mask]) ** 2)))

# ■■ Version 4: Streaming RMSE (for large datasets) ■■■■■■■■■■■■

class StreamingRMSE:

def __init__(self): self.n = self.sum_sq = 0

def update(self, actual: float, pred: float):

self.sum_sq += (actual - pred) ** 2

self.n += 1

@property

def value(self) -> float:

if self.n == 0: return float('nan')

return [Link](self.sum_sq / self.n)

# Usage: compute RMSE over streaming data without storing all values

rmse = StreamingRMSE()
for actual, pred in zip(actuals_stream, predictions_stream):

[Link](actual, pred)

print(f'RMSE: {[Link]:.4f}')

■ SAY THIS — Word for word

"I'd write it in three steps matching the formula: RMSE equals square root of mean of squared differences. For
production I'd use NumPy vectorised operations — convert to arrays, subtract element-wise, square, take mean,
square root. But I'd also show I understand edge cases: validate equal lengths, handle empty inputs, handle
NaN values with a mask. If the dataset is too large to fit in memory, I'd use the streaming version — accumulate
sum of squared errors and count incrementally, compute RMSE at the end. This avoids loading all predictions
into RAM."
SECTION F — PHP, MYSQL & BACKEND (ApexPlanet Internship)

You worked with PHP and MySQL at ApexPlanet. What did you build and what did
Q8
you learn about scalable system design?

■ CONCEPT — What the interviewer is testing

What to say about the internship:

This is a defend-and-connect question. You need to: (1) Describe what you built, (2) Show you learned
transferable concepts, (3) Connect it to ML engineering.

PHP + MySQL concepts you must know:

• SQL queries: SELECT, JOIN, WHERE, GROUP BY, ORDER BY, LIMIT

• Indexes: B-tree index makes SELECT WHERE fast — O(log n) vs O(n) full scan

• ACID: Atomicity, Consistency, Isolation, Durability — why databases are reliable

• N+1 query problem: fetching parent then N children with N separate queries — fix with JOIN

• Connection pooling: reuse DB connections instead of opening new one per request

How this connects to ML engineering:

• Feature store: SQL database stores pre-computed features for training and serving

• Experiment metadata: MLflow uses SQLite or PostgreSQL to store run metadata

• Prediction logging: log every inference to MySQL for monitoring and retraining

• Scalable design: the same patterns (indexing, connection pooling, caching) apply to ML serving databases

■ CODE — SQL patterns every ML engineer must know

-- ■■ Feature lookup for ML serving ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■

SELECT f.user_id, f.avg_purchase, f.days_since_last_order,

f.total_spend, [Link]

FROM features f

JOIN users u ON f.user_id = [Link]

WHERE f.updated_at > DATE_SUB(NOW(), INTERVAL 1 DAY) -- fresh features only

ORDER BY f.user_id;
-- ■■ Aggregation for feature engineering ■■■■■■■■■■■■■■■■■■■■■■

SELECT user_id,

COUNT(*) AS purchase_count,

AVG(amount) AS avg_order_value,

MAX(created_at) AS last_purchase,

DATEDIFF(NOW(), MAX(created_at)) AS recency_days

FROM orders

WHERE created_at > '2023-01-01'

GROUP BY user_id

HAVING purchase_count > 2; -- only customers with 2+ orders

-- ■■ Index for fast feature lookup ■■■■■■■■■■■■■■■■■■■■■■■■■■■

CREATE INDEX idx_user_id ON features(user_id);

CREATE INDEX idx_updated ON features(updated_at);

-- Without index: full table scan O(n). With index: O(log n).

-- ■■ Prediction logging (every inference logged) ■■■■■■■■■■■■■■

INSERT INTO predictions (

request_id, user_id, model_version,

input_hash, predicted_price, confidence, latency_ms

) VALUES (?, ?, ?, ?, ?, ?, ?);

-- ■■ Monitor model performance (actual vs predicted) ■■■■■■■■■■

SELECT model_version,

AVG(ABS(actual_price - predicted_price)) AS mae,

SQRT(AVG(POW(actual_price - predicted_price, 2))) AS rmse

FROM predictions p

JOIN actual_sales s ON p.request_id = s.request_id

WHERE p.created_at > DATE_SUB(NOW(), INTERVAL 7 DAY)

GROUP BY model_version;
■ SAY THIS — Word for word

"At ApexPlanet I built backend features in PHP with MySQL — form processing, user authentication, CRUD
operations for a web application. The most valuable thing I learned was thinking about database performance:
we had a page that was running 50+ queries per page load due to N+1 problems — fetching each user's order
history in a loop. I refactored it to a single JOIN query and added an index on user_id, which cut page load from
3 seconds to 200ms. This directly maps to ML engineering: I log every prediction to a MySQL table, and without
a proper index on model_version and created_at, my monitoring query that computes weekly RMSE would be a
full table scan on millions of rows. Same problem, same fix."
SESSION 8 — MASTER QUICK REFERENCE
Topic One-line core answer Key fact / code

Allows only 1 thread to run Python bytecode at a time. Bypass: multiprocessing

GIL
Released during I/O and C extensions. (Gunicorn)

Threads vs Processes vs Async: I/O-bound single thread. Threads: I/O with blocking FastAPI: async. Model:
Async libs. Multiprocessing: CPU-bound. multiprocessing.

Reference counting primary. Cyclic GC for circular

Python GC Deterministic unlike Java GC
references. del decrements refcount.

Higher-order function wrapping another. @[Link] func = decorator(func)

Decorator
preserves metadata. equivalent

Decorator with args: outer function returns decorator. @retry(max_attempts=3,

@retry decorator
Retries on exception with delay. delay=0.5)

Generator: lazy, one value at a time, O(1) memory. List:

Generator vs List 1M items: list=8MB, gen=200B
eager, all in memory, indexable.

Delegates to sub-generator — flattens nested generators.

yield from yield from load_chunks(file)
Cleaner than for x in gen: yield x.

ASGI async, Pydantic v2 (Rust validation), Starlette base, Pydantic v2: 5-50x faster than
FastAPI speed
dependency injection. v1

@asynccontextmanager lifespan: before yield=startup [Link] — never load

Lifespan events
(load model), after yield=shutdown. per request

Depends(func): FastAPI calls func before handler, injects Override in tests:

Dependency injection
return. Testable, reusable. app.dependency_overrides

Middleware: every request, cross-cutting (logging, CORS). Both used together in

Middleware vs Dependency
Dependency: specific routes. production

Nginx → FastAPI → Redis cache → ThreadPool inference Async queue (Celery) for > 2s
ML system design
→ CloudWatch logging → ASG scaling inference

New model runs in parallel with prod, predictions logged Before promoting to Production
Shadow mode
not served. Zero-risk validation. stage

ALB traffic split or user_id hash routing. Log model version Gradual: 5% → 20% → 50% →
A/B testing models
per request. Stats significance test. 100%

O(1) lookup for feature encoding, frequency counting, Counter, defaultdict from
Hash map pattern
groupby aggregation. collections

Rolling average/stats in O(n): add new, subtract old. Avoid

Sliding window window_sum += new - old
O(n×w) naive approach.

sqrt(mean((actuals - preds)²)). NumPy vectorised. Handle

RMSE from scratch [Link]([Link]((a-p)**2))
NaN mask. Streaming for large data.

Fetching N children with N separate queries. Fix: JOIN in

SQL N+1 problem 3s → 200ms with JOIN + index
one query + index on foreign key.
 Feature store queries, prediction logging, monitoring Index on user_id,
SQL for ML
RMSE/MAE over time via GROUP BY. model_version, date

ALL 8 SESSIONS COMPLETE — You now have a comprehensive interview preparation guide covering every
topic on your resume.