Python + Coding

Deep Dive for Data Engineers

Focused preparation for the Python live coding round at Tier A/S DE interviews. Covers
Python fundamentals interviewers probe, Pandas/PyArrow/Polars decision-making,
PySpark idioms beyond surface, 30 canonical coding problems with solutions, testing
patterns, and live coding round tactics.

What this document is

Type Topic-focused interview deep dive

Scope Python fluency for DE roles + 30 coding problems

Tier A/S quality - tested at Stripe, Atlassian, Datadog, Databricks,

Depth
FAANG

Audience 7-YOE DE who codes daily but hasn't focused on Python interview prep

Length ~60 pages, dense, exam-focused

Companion Use alongside System Design Playbook and Snowflake Deep Dive

How to use this document:

Part 4 (30 coding problems) is the highest-value section. For each problem, attempt it yourself with a
20-minute timer BEFORE reading the solution. Reading solutions teaches you nothing; struggling with the
problem first and then comparing to the solution teaches everything.
Generated May 27, 2026 | Personal reference

Python + Coding Deep Dive Page 1

 Table of contents
Part Topic Pages

1 Python fundamentals interviewers probe 10

2 Pandas / PyArrow / Polars - the decision 8

3 PySpark idioms beyond surface 10

4 30 canonical coding problems with solutions 22

5 Testing data pipelines 5

6 Live coding round tactics 5

Python + Coding Deep Dive Page 2

 PART 1
Python Fundamentals Interviewers Probe

Why this part matters

Senior DEs are assumed to know Python syntax. What interviewers test is whether you understand Python
deeply enough to write production-grade code: idioms that catch the experienced eye, gotchas that catch
the inexperienced one.

1.1 Mutable default arguments - the classic gotcha

# WRONG - the default list is shared across calls
def add_event(event, events=[]):
[Link](event)
return events

print(add_event('click')) # ['click']
print(add_event('view')) # ['click', 'view'] -- WTF?
# Default values are evaluated ONCE at function definition time
# All calls share the same list

# CORRECT
def add_event(event, events=None):
if events is None:
events = []
[Link](event)
return events

# Same applies to dicts, sets, other mutable defaults

# Interviewers ask this to see if you've been bitten by it in production

1.2 List comprehensions vs generators vs map/filter

Python + Coding Deep Dive Page 3

 # List comprehension - eager, materializes full list in memory
squares = [x**2 for x in range(1_000_000)] # 8 MB in memory

# Generator expression - lazy, yields one at a time

squares_gen = (x**2 for x in range(1_000_000)) # ~200 bytes

# When to use each:

# - List: when you need it multiple times, or to index into it
# - Generator: for streaming through data once (memory-bounded)

# For data engineering specifically: generators are the right default

# for pipeline stages because data sizes are unpredictable

# Chaining generators (pipeline-style):

def read_lines(path):
with open(path) as f:
for line in f:
yield [Link]()

def parse_json(lines):
import json
for line in lines:
try:
yield [Link](line)
except [Link]:
continue

def filter_events(records, event_type):

for r in records:
if [Link]('event_type') == event_type:
yield r

# Compose - nothing executes yet (lazy)

pipeline = filter_events(parse_json(read_lines('[Link]')), 'purchase')

# Now iterate - one record flows through entire pipeline at a time

for record in pipeline:
process(record)
# Memory usage: O(1) regardless of file size

1.3 Decorators - the patterns that actually matter

Python + Coding Deep Dive Page 4

 # Retry decorator - constant pattern in data pipelines
import functools, time

def retry(max_attempts=3, backoff_seconds=1):

def decorator(func):
@[Link](func)
def wrapper(*args, **kwargs):
attempt = 0
while attempt < max_attempts:
try:
return func(*args, **kwargs)
except Exception as e:
attempt += 1
if attempt == max_attempts:
raise
[Link](backoff_seconds * (2 ** attempt)) # exponential
return None
return wrapper
return decorator

@retry(max_attempts=5, backoff_seconds=2)
def fetch_api_data(url):
response = [Link](url, timeout=10)
response.raise_for_status()
return [Link]()

# Timing decorator - useful for instrumenting pipelines

def timed(func):
@[Link](func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f"{func.__name__} took {elapsed:.2f}s")
return result
return wrapper

# Caching with built-in functools.lru_cache

@functools.lru_cache(maxsize=1000)
def expensive_lookup(key):
return [Link](key)
# Memoizes; second call with same key is instant

Python + Coding Deep Dive Page 5

 1.4 Context managers - resource management done
right
# Built-in usage - file handling
with open('[Link]') as f:
data = [Link]()
# File auto-closes even if exception raised

# Custom context manager via class

class DatabaseConnection:
def __init__(self, conn_string):
self.conn_string = conn_string
[Link] = None

def __enter__(self):
[Link] = create_connection(self.conn_string)
return [Link]

def __exit__(self, exc_type, exc_val, exc_tb):

if [Link]:
[Link]()
return False # propagate exceptions

with DatabaseConnection('postgres://...') as conn:

result = [Link]('SELECT ...')
# Connection guaranteed to close

# Custom context manager via decorator (more Pythonic for simple cases)
from contextlib import contextmanager

@contextmanager
def temp_directory():
import tempfile, shutil
path = [Link]()
try:
yield path
finally:
[Link](path)

with temp_directory() as tmp:

# use tmp
pass
# Directory cleaned up automatically

# Multiple context managers in one with statement

with open('[Link]') as fin, open('[Link]', 'w') as fout:
for line in fin:
[Link](transform(line))

1.5 Type hints - production-grade Python

Python + Coding Deep Dive Page 6

 # Without type hints - works but unclear
def process(data, threshold=10):
return [x for x in data if x > threshold]

# With type hints - self-documenting; catches errors via mypy

from typing import List, Optional, Dict, Iterable, Callable

def process(data: List[int], threshold: int = 10) -> List[int]:

return [x for x in data if x > threshold]

# Type hints for data structures

def parse_event(raw: str) -> Optional[Dict[str, str]]:
try:
return [Link](raw)
except [Link]:
return None

# Generic types
def get_first(items: Iterable[int]) -> Optional[int]:
for item in items:
return item
return None

# Modern syntax (Python 3.10+)

def process(data: list[int], threshold: int = 10) -> list[int]:
return [x for x in data if x > threshold]

# Why interviewers care:

# - Type hints reduce production bugs
# - Enable IDE autocomplete and refactoring
# - Make APIs self-documenting
# - mypy --strict catches issues before production
# Senior code is type-hinted; junior code often isn't

Python + Coding Deep Dive Page 7

 1.6 Exception handling for pipelines
# WRONG - swallowing exceptions silently
try:
process(record)
except: # bare except catches EVERYTHING incl. KeyboardInterrupt
pass # silent failure - records lost without trace

# WRONG - too broad

try:
process(record)
except Exception:
pass # swallows real bugs

# BETTER - specific exceptions, logged

try:
process(record)
except [Link] as e:
[Link](f"Bad JSON, skipping: {record[:100]}, error: {e}")
[Link]('json_decode_errors')
except KeyError as e:
[Link](f"Missing field: {e}")
dead_letter_queue.put(record)
# Other exceptions propagate - real bugs surface

# Pipeline pattern: continue on error, but track

def process_batch(records):
results = []
errors = []
for record in records:
try:
[Link](transform(record))
except ([Link], ValidationError) as e:
[Link]({'record': record, 'error': str(e)})
return results, errors

# In Airflow / batch jobs:

results, errors = process_batch(records)
if len(errors) > len(records) * 0.05:
raise PipelineError(f"Error rate {len(errors)/len(records)} exceeds 5%")
# Tolerate small errors; fail hard on systemic issues

1.7 Async/await for I/O-heavy DE workflows

Python + Coding Deep Dive Page 8

 # Synchronous - 100 API calls serially = 100s if each takes 1s
import requests
def fetch_all_sync(urls):
return [[Link](url).json() for url in urls]

# Async - 100 concurrent calls = ~1s total

import asyncio, aiohttp

async def fetch_one(session, url):

async with [Link](url) as response:
return await [Link]()

async def fetch_all_async(urls):

async with [Link]() as session:
tasks = [fetch_one(session, url) for url in urls]
return await [Link](*tasks)

results = [Link](fetch_all_async(urls))

# When to use async in DE:

# - Fetching from many APIs (enrichment)
# - Reading from many files concurrently
# - Database operations with multiple connections
# DON'T use for: CPU-bound work (use multiprocessing instead)

# Concurrency limit - avoid overwhelming downstream

async def fetch_all_limited(urls, max_concurrent=10):
sem = [Link](max_concurrent)
async with [Link]() as session:
async def bounded_fetch(url):
async with sem:
return await fetch_one(session, url)
tasks = [bounded_fetch(url) for url in urls]
return await [Link](*tasks)

Python + Coding Deep Dive Page 9

 1.8 Dataclasses - the modern way to model data
# Old style - lots of boilerplate
class Event:
def __init__(self, event_id, user_id, timestamp, properties=None):
self.event_id = event_id
self.user_id = user_id
[Link] = timestamp
[Link] = properties or {}

def __repr__(self):
return f"Event({self.event_id}, {self.user_id})"

def __eq__(self, other):

return self.event_id == other.event_id

# Modern style with dataclasses

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, Any

@dataclass
class Event:
event_id: str
user_id: str
timestamp: datetime
properties: Dict[str, Any] = field(default_factory=dict)

# Auto-generated: __init__, __repr__, __eq__

# Frozen for immutability:
@dataclass(frozen=True)
class ImmutableEvent:
event_id: str
user_id: str

# Use in pipelines:
events = [Event(e['id'], e['user_id'], e['ts']) for e in raw_events]

# Pydantic for validation (preferred in many modern pipelines)

from pydantic import BaseModel, validator

class Event(BaseModel):
event_id: str
user_id: str
amount: float

@validator('amount')
def amount_positive(cls, v):
if v <= 0:
raise ValueError('amount must be positive')
return v

# Validates on construction; catches bad data early

1.9 Iterators and the data engineering mindset

Python + Coding Deep Dive Page 10

 # Custom iterator for chunked file processing
class ChunkedReader:
def __init__(self, file_path, chunk_size=10000):
self.file_path = file_path
self.chunk_size = chunk_size

def __iter__(self):
with open(self.file_path) as f:
chunk = []
for line in f:
[Link]([Link]())
if len(chunk) >= self.chunk_size:
yield chunk
chunk = []
if chunk:
yield chunk

# Process huge files in bounded memory

for chunk in ChunkedReader('huge_file.txt', chunk_size=10000):
process_chunk(chunk) # only 10K lines in memory at any time

# itertools - the data engineering swiss army knife

from itertools import (
islice, # take first N from any iterable
chain, # concatenate multiple iterables lazily
groupby, # group consecutive equal elements
takewhile, # take while condition true
dropwhile, # skip while condition true
accumulate, # running totals
)

# Take first 1000 records from a generator

first_1000 = list(islice(infinite_generator, 1000))

# Group by date (assumes data is sorted)

for date, group in groupby(sorted_events, key=lambda e: e['date']):
daily_count = sum(1 for _ in group)
print(f"{date}: {daily_count}")

Python + Coding Deep Dive Page 11

 1.10 Python performance - what you should know
# Profiling - find bottlenecks before optimizing
import cProfile, pstats

def my_pipeline():
# ...
pass

[Link]('my_pipeline()', '[Link]')
stats = [Link]('[Link]')
stats.sort_stats('cumulative').print_stats(20)

# Common performance wins:

# 1. Use built-ins (implemented in C)

# Slow:
total = 0
for x in data: total += x
# Fast:
total = sum(data)

# 2. Avoid global variable lookups in tight loops

import math
# Slow: [Link] looked up every iteration
results = [[Link](x) for x in data]
# Fast: local binding
sqrt = [Link]
results = [sqrt(x) for x in data]

# 3. Use sets/dicts for membership checks

# Slow: O(n) per check
items_list = [...] # 100K items
if x in items_list: ...
# Fast: O(1) per check
items_set = set(items_list)
if x in items_set: ...

# 4. String concatenation in loops

# Slow: each += creates new string
result = ''
for s in strings:
result += s
# Fast:
result = ''.join(strings)

# 5. Use array operations for numeric work

# Slow: pure Python loop
result = [x * 2 + 5 for x in numbers]
# Fast: NumPy vectorized
import numpy as np
arr = [Link](numbers)
result = arr * 2 + 5 # 10-100x faster for large arrays

# 6. GIL awareness
# Python's Global Interpreter Lock means threading helps with I/O,
# but NOT CPU-bound work. For CPU-bound, use multiprocessing.
from multiprocessing import Pool
with Pool(8) as pool:
results = [Link](heavy_cpu_function, data)

Python + Coding Deep Dive Page 12

 Interview angle

Common Tier A/S question: 'Show me how you'd process a 10GB CSV that doesn't fit in memory.'
Answer: 'Use a generator-based pipeline. Read line by line OR in chunks via pandas read_csv with
chunksize. Process each chunk, write output incrementally. For structured data, PyArrow's
incremental reader is even better - memory-mapped, zero-copy. If processing is CPU-bound,
parallelize across chunks with [Link]. If I/O-bound (e.g., calling APIs per row), use
[Link] with a semaphore.' Mentioning generators, PyArrow, and the CPU vs I/O distinction
signals depth.

Python + Coding Deep Dive Page 13

 PART 2
Pandas / PyArrow / Polars - The Decision

2.1 The decision tree

Three libraries for in-memory data work. Knowing when to reach for each separates Senior from Staff
candidates.

Data size Best choice Why

< 1M rows / < 1 GB Pandas Maturity; ecosystem; team familiarity

1M - 100M rows / 1-10 GB Polars 2-10x faster than Pandas; lower memory

> 100M rows / > 10 GB PyArrow + Spark Memory-mapped; out-of-core; distributed if needed

Storage / serialization PyArrow Parquet, Arrow IPC, zero-copy

Streaming / chunked Pandas (chunksize) or Bounded memory

Polars (lazy)

2.2 Pandas - the essentials interviewers test

Python + Coding Deep Dive Page 14

 import pandas as pd
import numpy as np

# Read large CSV in chunks (bounded memory)

chunks = pd.read_csv('[Link]', chunksize=100_000)
results = []
for chunk in chunks:
processed = chunk[chunk['amount'] > 0].groupby('user_id').agg({'amount': 'sum'})
[Link](processed)
final = [Link](results).groupby(level=0).sum()

# Read Parquet (much faster + smaller than CSV)

df = pd.read_parquet('[Link]', columns=['user_id', 'amount']) # column pruning

# Common patterns
# Window functions
df['rank'] = [Link]('user_id')['amount'].rank(method='dense', ascending=False)
df['cumsum'] = [Link]('user_id')['amount'].cumsum()
df['rolling_7d'] = [Link]('user_id')['amount'].rolling(7).mean().reset_index(0, drop=True)

# Pivot table
pivoted = df.pivot_table(
index='user_id',
columns='product_category',
values='amount',
aggfunc='sum',
fill_value=0
)

# Merge (Pandas' join)

result = [Link](customers, on='customer_id', how='left')
# how options: 'inner', 'left', 'right', 'outer'

# Filtering with .query() - more readable for complex filters

high_value = [Link]("amount > 100 and status == 'paid'")

# Method chaining for readability

result = (df
.query("status == 'completed'")
.assign(revenue=lambda x: x['quantity'] * x['price'])
.groupby('customer_id')
.agg({'revenue': 'sum', 'order_id': 'count'})
.rename(columns={'order_id': 'order_count'})
.sort_values('revenue', ascending=False)
.head(100)
)

Pandas common performance pitfalls

Python + Coding Deep Dive Page 15

 # SLOW: iterating with .iterrows()
for idx, row in [Link](): # 50x slower than vectorized
[Link][idx, 'doubled'] = row['x'] * 2

# FAST: vectorized
df['doubled'] = df['x'] * 2

# SLOW: .apply() with axis=1

df['result'] = [Link](lambda row: row['a'] + row['b'], axis=1)

# FAST: vectorized
df['result'] = df['a'] + df['b']

# SLOW: repeated column access in loop

for i in range(len(df)):
val = df['col'].iloc[i] # repeated lookup

# FAST: convert to numpy array

arr = df['col'].to_numpy()
for i in range(len(arr)):
val = arr[i]

# SLOW: filtering with chained []

df[df['a'] > 0][df['b'] > 0] # creates intermediate df

# FAST: single filter

df[(df['a'] > 0) & (df['b'] > 0)]

Python + Coding Deep Dive Page 16

 2.3 Polars - the modern alternative
Polars is a Rust-based DataFrame library with a Pandas-like API but 2-10x faster on most operations,
lower memory usage, and lazy evaluation support.

import polars as pl

# Read CSV (eager)

df = pl.read_csv('[Link]')

# Read CSV (lazy - operations queue up; execute on collect())

df = pl.scan_csv('[Link]') # returns LazyFrame
result = (df
.filter([Link]('amount') > 100)
.group_by('user_id')
.agg([Link]('amount').sum())
.collect() # actually executes
)

# Why lazy matters:

# Polars query optimizer reorders, prunes columns, pushes filters down
# Often 5-10x faster than eager equivalent

# Read Parquet with column + row pruning (super fast)

result = (pl.scan_parquet('[Link]')
.select(['user_id', 'amount', 'order_date']) # column pruning
.filter([Link]('order_date') >= '2026-01-01') # row pruning
.collect()
)
# Only reads needed columns + rows from disk - even for huge files

# Window functions
df = df.with_columns([
[Link]('amount').rank('dense', descending=True).over('user_id').alias('rank'),
[Link]('amount').cumsum().over('user_id').alias('cumsum'),
])

# Pandas <-> Polars interop (zero-copy via Arrow)

import pandas as pd
pandas_df = df.to_pandas()
polars_df = pl.from_pandas(pandas_df)

When Polars wins over Pandas

Data size 1-100GB: Polars handles this band much better. Pandas struggles >10GB; Polars handles
routinely.
Multi-column operations: Polars's expression API parallelizes; Pandas is single-threaded.
Lazy evaluation: Optimization opportunities Pandas can't do.
Type safety: Polars is strict about types; catches errors earlier.
When NOT to switch: if your team and tooling are Pandas-centric, switching cost may exceed
performance gain. Pandas ecosystem (matplotlib, scikit-learn) is still bigger.

Killer detail

Mentioning Polars by name in interviews signals you're current with Python data ecosystem. Most
candidates default to 'I'd use Pandas'. Saying 'For this size I'd actually reach for Polars - lazy
evaluation lets the optimizer push filters down to the Parquet read' is a clear differentiator.

Python + Coding Deep Dive Page 17

 2.4 PyArrow - the underlying foundation
PyArrow is the Apache Arrow Python binding. Both Pandas (modern versions) and Polars use PyArrow
under the hood. Worth knowing directly for storage and zero-copy interop.

import pyarrow as pa
import [Link] as pq

# Read Parquet directly with PyArrow

table = pq.read_table('[Link]')
# table is a [Link] - columnar in-memory

# Convert to Pandas (potentially zero-copy)

df = table.to_pandas()

# Read with column pruning + filter pushdown

table = pq.read_table(
'[Link]',
columns=['user_id', 'amount'],
filters=[('order_date', '>=', '2026-01-01')] # pushed to Parquet reader
)

# Write Parquet with control over compression + row groups

pq.write_table(
table,
'[Link]',
compression='snappy', # snappy, gzip, zstd, brotli
row_group_size=100_000 # rows per row group
)

# Stream large files (don't load all at once)

parquet_file = [Link]('[Link]')
for batch in parquet_file.iter_batches(batch_size=10_000):
# process each RecordBatch
df_chunk = batch.to_pandas()
process(df_chunk)

# Arrow IPC - inter-process / inter-language data exchange

with [Link]('[Link]', 'wb') as sink:
with [Link].new_file(sink, [Link]) as writer:
writer.write_table(table)

# Read back (zero-copy)

with pa.memory_map('[Link]') as source:
table = [Link].open_file(source).read_all()
# Memory-mapped = OS handles paging; can be much bigger than RAM

Why PyArrow matters for DE

1. Zero-copy interop. Pandas -> Polars -> Spark all via Arrow without serialization cost.
2. Memory-mapped reads. Process files much bigger than RAM.
3. Predicate + projection pushdown. Parquet reader only reads needed columns and rows.
4. The format underlying Iceberg/Delta/Hudi. Knowing Arrow = knowing the storage layer better.
5. The Snowflake Snowpark and BigQuery client both use Arrow for query result transfer.

2.5 Mental model: choose the tool

Python + Coding Deep Dive Page 18

 # Decision flowchart:

if data_size < 1_GB and team_uses_pandas:

use_pandas()
elif data_size < 100_GB:
use_polars() # major perf gain; mostly Pandas-compatible API
elif distributed_compute_available:
use_spark() # PySpark for big data
else:
use_pyarrow_streaming() # bounded memory on single machine

# Storage:
# - For Parquet/Arrow I/O: PyArrow directly (most control)
# - For quick local files: Pandas/Polars wrappers (convenience)

# Interop:
# - Always use Arrow as the lingua franca between systems
# - Avoid CSV in production (slow, lossy, ambiguous types)
# - Prefer Parquet for analytics; JSON for events/logs

Python + Coding Deep Dive Page 19

 PART 3
PySpark Idioms Beyond Surface

3.1 The DataFrame API patterns

from [Link] import SparkSession, functions as F, Window

spark = [Link]('my_pipeline').getOrCreate()

# Read with schema (faster than inference; production preferred)

from [Link] import StructType, StructField, StringType, LongType, DoubleType
schema = StructType([
StructField('order_id', LongType(), False),
StructField('user_id', StringType(), False),
StructField('amount', DoubleType(), True),
StructField('order_date', StringType(), False),
])
df = [Link](schema).parquet('s3://bucket/orders/')

# Filtering - prefer column expressions over SQL strings (better type checking)
filtered = [Link]([Link]('amount') > 100)
filtered = [Link](([Link]('amount') > 100) & ([Link]('status') == 'paid'))

# Selecting + transforming
result = [Link](
[Link]('order_id'),
[Link]('user_id'),
[Link]('amount'),
[Link]('order_date').alias('year'),
[Link]('order_date').alias('month'),
([Link]('amount') * 0.18).alias('tax')
)

# Aggregations
agg = [Link]('user_id').agg(
[Link]('*').alias('order_count'),
[Link]('amount').alias('total_spent'),
[Link]('amount').alias('avg_amount'),
[Link]('order_date').alias('last_order'),
[Link]('product_id').alias('distinct_products')
)

# Join
result = [Link](customers, on='customer_id', how='inner')
# Join types: 'inner', 'left', 'right', 'outer', 'left_semi', 'left_anti'

# Window functions
window_spec = [Link]('user_id').orderBy([Link]('order_date').desc())
df_with_rank = [Link]('rank', F.row_number().over(window_spec))

# Running totals
window_running = [Link]('user_id').orderBy('order_date') \
.rowsBetween([Link], [Link])
df = [Link]('cumulative_amount', [Link]('amount').over(window_running))

3.2 UDFs - and why to avoid them

Python + Coding Deep Dive Page 20

 # Plain Python UDF - SLOW (row-by-row Python execution)
@[Link]('double')
def calculate_discount(amount, tier):
if tier == 'gold':
return amount * 0.2
elif tier == 'silver':
return amount * 0.1
return 0.0

df = [Link]('discount', calculate_discount([Link]('amount'), [Link]('tier')))

# 10-100x slower than built-in equivalent

# Pandas UDF - FASTER (vectorized; works on Pandas Series via Arrow)

import pandas as pd

@F.pandas_udf('double')
def calculate_discount_pandas(amount: [Link], tier: [Link]) -> [Link]:
result = [Link](0.0, index=[Link])
result[tier == 'gold'] = amount[tier == 'gold'] * 0.2
result[tier == 'silver'] = amount[tier == 'silver'] * 0.1
return result
# 5-20x faster than plain UDF; uses Arrow for serialization

# BEST: built-in functions (no Python at all, runs in JVM/native)

df = [Link]('discount',
[Link]([Link]('tier') == 'gold', [Link]('amount') * 0.2)
.when([Link]('tier') == 'silver', [Link]('amount') * 0.1)
.otherwise(0.0)
)
# Fastest - executed natively in Spark engine

# RULE: Always check if a built-in exists before writing a UDF.

# Spark has 300+ built-in functions; usually one fits.

Python + Coding Deep Dive Page 21

 3.3 Broadcasting, repartitioning, coalescing
# Broadcast join - small table replicated to all executors
# Avoids shuffle of large table
from [Link] import broadcast

large_df.join(broadcast(small_df), on='key')
# Spark auto-broadcasts tables < [Link] (10MB default)
# Explicit broadcast() forces it for slightly larger tables

# Repartition - increases partitions, forces shuffle

[Link](200) # 200 partitions, random distribution
[Link]('user_id') # hash partition by user_id
[Link](200, 'user_id') # both
# Use BEFORE expensive joins/aggregations to pre-distribute by key

# Coalesce - decreases partitions, NO shuffle

[Link](10) # 10 partitions, no rebalancing
# Use AFTER aggregations when result is small (avoid 200 tiny output files)

# Common pattern: aggregate then coalesce

agg = large_df.groupBy('user_id').agg([Link]('amount'))
[Link](50).[Link]('output/') # 50 reasonably-sized output files

# Salt the hot key for skew (covered in Spark study notes)
# Two-stage: salt -> partial agg -> de-salt -> final agg
df_salted = [Link]('salted_key',
[Link]([Link]('key'), [Link]('_'), ([Link]() * 100).cast('int')))
# ... aggregate by salted_key
# ... strip salt
# ... final aggregate by key

3.4 Reading the explain plan

Python + Coding Deep Dive Page 22

 # See the physical plan
[Link]() # short plan
[Link](True) # detailed: parsed, analyzed, optimized, physical

# Example output to read:

# == Physical Plan ==
# *(2) HashAggregate(keys=[user_id], functions=[sum(amount)])
# +- Exchange hashpartitioning(user_id, 200)
# +- *(1) HashAggregate(keys=[user_id], functions=[partial_sum(amount)])
# +- FileScan parquet [user_id,amount]
# DataFilters: [...]
# PushedFilters: [IsNotNull(user_id)]

# What to look for:

# - "Exchange hashpartitioning" = shuffle (expensive)
# - "HashAggregate" with "partial_" prefix = pre-aggregation (good - reduces shuffle)
# - "FileScan parquet ... PushedFilters" = filter pushdown to Parquet reader (good)
# - "BroadcastHashJoin" = small table broadcast (efficient)
# - "SortMergeJoin" = large+large join, shuffle on both (expensive but unavoidable for huge)
# - "BroadcastNestedLoopJoin" = CROSS JOIN; usually a bug

# Optimization: ensure pushdowns happen

# Push filter BEFORE join (Spark usually does this; verify)
[Link]('amount > 100').join(other_df, 'key') # filter pushes down
[Link](other_df, 'key').filter('amount > 100') # also pushes down (Spark optimizes)

# Push filter to file scan level (best)

df = [Link]('data/').filter('order_date >= "2026-01-01"')
# Parquet reader skips entire row groups based on column statistics

Python + Coding Deep Dive Page 23

 3.5 Common PySpark interview problems
Problem: Top-N per group

# Top 3 orders per customer by amount

from [Link] import Window

w = [Link]('customer_id').orderBy([Link]('amount').desc())
top3 = (df
.withColumn('rank', F.row_number().over(w))
.filter([Link]('rank') <= 3)
.drop('rank')
)

# Alternative: use rank() if ties should be included

# Or dense_rank() for ranks without gaps

Problem: Sessionization

# Group user events into sessions (30-min inactivity = new session)

from [Link] import Window

w = [Link]('user_id').orderBy('event_time')

result = (events
.withColumn('prev_event_time', [Link]('event_time').over(w))
.withColumn('time_gap_seconds',
F.unix_timestamp('event_time') - F.unix_timestamp('prev_event_time'))
.withColumn('is_new_session',
[Link]([Link]('time_gap_seconds') > 1800, 1).otherwise(0))
.withColumn('session_id',
[Link]('is_new_session').over([Link]([Link], 0)))
)

# Now group by (user_id, session_id) to get per-session metrics

sessions = [Link]('user_id', 'session_id').agg(
[Link]('event_time').alias('session_start'),
[Link]('event_time').alias('session_end'),
[Link]('*').alias('event_count')
)

Problem: Deduplication with priority

Python + Coding Deep Dive Page 24

 # Keep only the latest record per (user_id, email) combination
w = [Link]('user_id', 'email').orderBy([Link]('updated_at').desc())

deduped = (df
.withColumn('rn', F.row_number().over(w))
.filter([Link]('rn') == 1)
.drop('rn')
)

# Alternative for cases where you want the MAX of one column but
# keep other columns intact - use group + struct trick:
result = (df
.groupBy('user_id', 'email')
.agg([Link]([Link]('updated_at', 'name', 'address', 'phone')).alias('latest'))
.select('user_id', 'email', 'latest.*')
)

Problem: Pivot a long-format table

# Long format: user_id, metric_name, value

# Wide format: user_id, metric1, metric2, metric3

wide = (long_df
.groupBy('user_id')
.pivot('metric_name', ['clicks', 'views', 'purchases']) # explicit values = faster
.agg([Link]('value'))
)

# Unpivot (Spark 3.4+): wide to long

long = [Link](
ids='user_id',
values=['clicks', 'views', 'purchases'],
variableColumnName='metric_name',
valueColumnName='value'
)

Python + Coding Deep Dive Page 25

 PART 4
30 Canonical Coding Problems

These 30 problems cover the patterns that come up in ~80% of Tier A/S DE coding rounds. For each:
problem statement, naive solution, optimal solution, complexity analysis, common follow-ups.

How to use: set a 20-minute timer and attempt each problem before reading the solution. Skipping this
step makes the doc useless. Struggle is where learning happens.

Categories: 10 data manipulation, 10 algorithm patterns, 10 PySpark-specific.

Section A: Data Manipulation Problems (1-10)

Problem 1: Rolling 7-day average per customer
Given: list of transactions [{'customer_id', 'date', 'amount'}, ...]. Return: for each (customer_id, date), the
rolling 7-day average amount.

Python + Coding Deep Dive Page 26

 # Naive O(N^2) approach: for each row, scan back 7 days
# Don't do this.

# Pandas solution - O(N log N) via sort + rolling

import pandas as pd

def rolling_7d_avg(transactions):
df = [Link](transactions)
df['date'] = pd.to_datetime(df['date'])
df = df.sort_values(['customer_id', 'date'])
df['rolling_avg'] = ([Link]('customer_id')
.rolling('7D', on='date')['amount']
.mean().reset_index(level=0, drop=True))
return df.to_dict('records')

# Pure Python solution (if no Pandas allowed)

from collections import defaultdict, deque
from datetime import datetime, timedelta

def rolling_7d_avg_pure(transactions):
# Group + sort by customer
by_customer = defaultdict(list)
for t in transactions:
by_customer[t['customer_id']].append(t)
for customer_id in by_customer:
by_customer[customer_id].sort(key=lambda x: x['date'])

result = []
for customer_id, txns in by_customer.items():
window = deque() # (date, amount) tuples in window
window_sum = 0.0
for t in txns:
current_date = [Link](t['date'])
# Remove items > 7 days old
while window and (current_date - window[0][0]).days > 7:
window_sum -= [Link]()[1]
[Link]((current_date, t['amount']))
window_sum += t['amount']
[Link]({**t, 'rolling_avg': window_sum / len(window)})
return result

# Follow-ups interviewers ask:

# - What if data isn't sorted? (sort first, mention O(N log N))
# - What if data doesn't fit in memory? (chunk by customer, process per group)
# - What's the rolling window's exact semantic? (last 7 days incl. current?
# or last 7 distinct days? Clarify before coding.)

Python + Coding Deep Dive Page 27

 Problem 2: Sessionization
Given: list of events [{'user_id', 'event_time', 'event_type'}, ...]. Return: list of sessions where a session is
consecutive events from the same user with <30 min gap.

from collections import defaultdict

from datetime import datetime, timedelta

def sessionize(events, gap_minutes=30):

# Group + sort by user
by_user = defaultdict(list)
for e in events:
by_user[e['user_id']].append(e)
for user_id in by_user:
by_user[user_id].sort(key=lambda x: x['event_time'])

sessions = []
for user_id, user_events in by_user.items():
current_session = []
last_time = None
for e in user_events:
t = [Link](e['event_time'])
if last_time and (t - last_time) > timedelta(minutes=gap_minutes):
# New session - emit current and start fresh
[Link]({
'user_id': user_id,
'start': current_session[0]['event_time'],
'end': current_session[-1]['event_time'],
'event_count': len(current_session),
})
current_session = []
current_session.append(e)
last_time = t
if current_session:
[Link]({
'user_id': user_id,
'start': current_session[0]['event_time'],
'end': current_session[-1]['event_time'],
'event_count': len(current_session),
})
return sessions

# Complexity: O(N log N) due to sort per user

# Space: O(N) for grouped events

# Follow-ups:
# - What if events come in unsorted from Kafka in real-time?
# (Use Flink session window or Spark Streaming windowing)
# - What if a user has 1M events? (Process incrementally; emit sessions as they close)
# - What if session ends are defined differently? (Logout event vs gap?
# Clarify; implement is_session_end() predicate)

Problem 3: Top K most frequent items in a stream

Given: stream of events. Return: top K most frequent event types at any time. Memory must be bounded.

Python + Coding Deep Dive Page 28

 # Approach 1: Exact - dict + heap, O(N) space (bad for unbounded stream)
from collections import Counter
import heapq

def top_k_exact(events, k):

counter = Counter(events)
return [Link](k, [Link](), key=lambda x: x[1])
# Memory grows with distinct events; not bounded.

# Approach 2: Count-Min Sketch (approximate, bounded memory)

class CountMinSketch:
def __init__(self, width=1000, depth=5):
[Link] = width
[Link] = depth
[Link] = [[0] * width for _ in range(depth)]
[Link] = [hash(f'seed_{i}') for i in range(depth)]

def add(self, item):

for i in range([Link]):
h = hash(([Link][i], item)) % [Link]
[Link][i][h] += 1

def estimate(self, item):

return min(
[Link][i][hash(([Link][i], item)) % [Link]]
for i in range([Link])
)

# Combine CMS with heap of candidate top-K

def top_k_streaming(stream, k, heavy_hitter_threshold=100):
cms = CountMinSketch()
heavy_hitters = {} # candidate top-K, bounded size
for item in stream:
[Link](item)
count = [Link](item)
if item in heavy_hitters:
heavy_hitters[item] = count
elif len(heavy_hitters) < k * 3: # keep 3K candidates
heavy_hitters[item] = count
elif count > min(heavy_hitters.values()):
# Replace least-frequent candidate
del heavy_hitters[min(heavy_hitters, key=heavy_hitters.get)]
heavy_hitters[item] = count
return [Link](k, heavy_hitters.items(), key=lambda x: x[1])

# Follow-ups:
# - How to handle streaming with windows (last hour only)?
# Use sliding CMS with multiple sub-sketches per time bucket
# - Tradeoff between accuracy and memory? width/depth controls error bound

Python + Coding Deep Dive Page 29

 Problem 4: Deduplicate by keeping latest per key
Given: records with (id, timestamp, ...other fields). Return: only the latest record per id.

# Approach 1: Sort and dedup - O(N log N)

def dedup_latest(records):
sorted_records = sorted(records, key=lambda r: (r['id'], r['timestamp']),
reverse=True)
seen = set()
result = []
for r in sorted_records:
if r['id'] not in seen:
[Link](r['id'])
[Link](r)
return result

# Approach 2: Single pass with dict - O(N)

def dedup_latest_v2(records):
latest = {}
for r in records:
if r['id'] not in latest or r['timestamp'] > latest[r['id']]['timestamp']:
latest[r['id']] = r
return list([Link]())

# Pandas equivalent
import pandas as pd
def dedup_latest_pandas(records):
df = [Link](records)
return (df.sort_values('timestamp', ascending=False)
.drop_duplicates(subset='id', keep='first')
.to_dict('records'))

# Spark equivalent
from [Link] import Window
import [Link] as F
def dedup_latest_spark(df):
w = [Link]('id').orderBy([Link]('timestamp').desc())
return ([Link]('rn', F.row_number().over(w))
.filter([Link]('rn') == 1).drop('rn'))

# Follow-up: what if ties on timestamp?

# Clarify: most recent insertion order? Specific tiebreaker column?

Problem 5: Merge two sorted lists / streams

Given: two sorted lists of events. Return: merged sorted list. Standard interview problem; pattern for
merging any number of sorted streams.

Python + Coding Deep Dive Page 30

 def merge_two(a, b, key=lambda x: x):
i = j = 0
result = []
while i < len(a) and j < len(b):
if key(a[i]) <= key(b[j]):
[Link](a[i])
i += 1
else:
[Link](b[j])
j += 1
[Link](a[i:])
[Link](b[j:])
return result

# Merging K sorted streams - heap-based

import heapq

def merge_k_sorted(streams, key=lambda x: x):

# Each stream is an iterator
# Use a heap of (key, stream_index, value, next_value_from_stream)
heap = []
iterators = [iter(s) for s in streams]
for i, it in enumerate(iterators):
first = next(it, None)
if first is not None:
[Link](heap, (key(first), i, first))

while heap:
k, i, val = [Link](heap)
yield val
next_val = next(iterators[i], None)
if next_val is not None:
[Link](heap, (key(next_val), i, next_val))

# Time complexity: O(N log K) where N = total items, K = streams

# Space: O(K)
# This is the pattern behind Kafka consumer group multi-partition reads,
# Iceberg manifest merging, Snowflake metadata aggregation

Python + Coding Deep Dive Page 31

 Problem 6: Find missing date ranges (gaps)
Given: list of date ranges [(start, end), ...] and an overall window [overall_start, overall_end]. Return: the
gaps - dates in overall window not covered by any range.

from datetime import date, timedelta

def find_gaps(ranges, overall_start, overall_end):

if not ranges:
return [(overall_start, overall_end)]

# Sort by start
ranges = sorted(ranges, key=lambda r: r[0])

# Merge overlapping
merged = [ranges[0]]
for start, end in ranges[1:]:
last_start, last_end = merged[-1]
if start <= last_end + timedelta(days=1):
# Overlap or adjacent - extend
merged[-1] = (last_start, max(last_end, end))
else:
[Link]((start, end))

# Find gaps
gaps = []
cursor = overall_start
for start, end in merged:
if cursor < start:
[Link]((cursor, start - timedelta(days=1)))
cursor = max(cursor, end + timedelta(days=1))
if cursor <= overall_end:
[Link]((cursor, overall_end))

return gaps

# Test:
# ranges = [(date(2026,1,1), date(2026,1,5)), (date(2026,1,10), date(2026,1,15))]
# overall = (date(2026,1,1), date(2026,1,31))
# Returns: [(date(2026,1,6), date(2026,1,9)), (date(2026,1,16), date(2026,1,31))]

# Use case: detecting missing data days in a pipeline

# Use case: finding holes in time-series for backfill jobs

Problem 7: Cohort analysis - user retention

Given: list of (user_id, signup_date, active_date) records. Return: cohort retention - for each
signup_month, what % of users were active in month 1, 2, 3, ... after signup.

Python + Coding Deep Dive Page 32

 import pandas as pd

def cohort_retention(activity_data):
df = [Link](activity_data)
df['signup_date'] = pd.to_datetime(df['signup_date'])
df['active_date'] = pd.to_datetime(df['active_date'])

df['signup_month'] = df['signup_date'].dt.to_period('M')
df['active_month'] = df['active_date'].dt.to_period('M')
df['months_after_signup'] = (
(df['active_month'] - df['signup_month']).apply(lambda x: x.n)
)

# Pivot: rows = signup_month, columns = months_after, values = unique users

cohort = ([Link](['signup_month', 'months_after_signup'])['user_id']
.nunique()
.unstack(fill_value=0))

# Convert to retention % (divide by cohort size at month 0)

cohort_size = cohort[0] # month 0 = signup month, all active
retention = [Link](cohort_size, axis=0)

return retention

# Output looks like:

# 0 1 2 3 4
# signup_month
# 2025-09 1.00 0.65 0.48 0.40 0.35
# 2025-10 1.00 0.62 0.50 0.42 ...
# 2025-11 1.00 0.68 0.52 ... ...

# This is THE canonical analytics engineering problem

Python + Coding Deep Dive Page 33

 Problem 8: Find first event of type X after event Y
Given: sequence of user events. For each 'click_ad' event, find the next 'purchase' event by the same
user (if any) within 24 hours. This is attribution analysis.

from collections import defaultdict

from datetime import datetime, timedelta

def attribute_clicks_to_purchases(events, window_hours=24):

by_user = defaultdict(list)
for e in events:
by_user[e['user_id']].append(e)
for user_id in by_user:
by_user[user_id].sort(key=lambda x: x['event_time'])

attributions = []
for user_id, user_events in by_user.items():
i = 0
while i < len(user_events):
if user_events[i]['event_type'] != 'click_ad':
i += 1
continue
click = user_events[i]
click_time = [Link](click['event_time'])
window_end = click_time + timedelta(hours=window_hours)

# Find next purchase within window

for j in range(i + 1, len(user_events)):
e = user_events[j]
if [Link](e['event_time']) > window_end:
break
if e['event_type'] == 'purchase':
[Link]({
'user_id': user_id,
'click_time': click['event_time'],
'purchase_time': e['event_time'],
'attributed_revenue': e['amount'],
})
break
i += 1
return attributions

# Spark equivalent uses temporal join (AS OF JOIN if Spark version supports)
# Or window function with lead()

# Follow-up: what about multi-touch attribution?

# (Multiple clicks lead to one purchase; distribute credit across clicks)
# Clarify the attribution model with interviewer (first-touch, last-touch,
# linear, time-decay)

Problem 9: Anomaly detection - z-score per group

Given: metrics by (date, region). Return: anomalies where the value is >3 standard deviations from the
28-day rolling mean for that region.

Python + Coding Deep Dive Page 34

 import pandas as pd
import numpy as np

def detect_anomalies(metrics, window_days=28, z_threshold=3.0):

df = [Link](metrics)
df['date'] = pd.to_datetime(df['date'])
df = df.sort_values(['region', 'date'])

# Per-region rolling stats

df['rolling_mean'] = ([Link]('region')['value']
.rolling(window_days, min_periods=7).mean()
.reset_index(level=0, drop=True))
df['rolling_std'] = ([Link]('region')['value']
.rolling(window_days, min_periods=7).std()
.reset_index(level=0, drop=True))

df['z_score'] = (df['value'] - df['rolling_mean']) / df['rolling_std']

df['is_anomaly'] = df['z_score'].abs() > z_threshold

return df[df['is_anomaly']].to_dict('records')

# Production version (Spark for large scale)

import [Link] as F
from [Link] import Window

def detect_anomalies_spark(df, window_days=28, z_threshold=3.0):

w = [Link]('region').orderBy('date') \
.rowsBetween(-window_days, -1) # past N days, excl. current

return (df
.withColumn('rolling_mean', [Link]('value').over(w))
.withColumn('rolling_std', [Link]('value').over(w))
.withColumn('z_score',
([Link]('value') - [Link]('rolling_mean')) / [Link]('rolling_std'))
.filter([Link]('z_score') > z_threshold)
)

# Follow-ups:
# - What if there are 0 std (constant series)? Handle div-by-zero
# - What about seasonality (e.g., weekly patterns)? Use STL decomposition first
# - What's a 'real' anomaly vs noise? Combine z-score with MAD; use multiple signals

Python + Coding Deep Dive Page 35

 Problem 10: Sliding window aggregation in real-time
Given: stream of (timestamp, value) tuples. Return: running stats (count, sum, mean) over the last 5
minutes as new events arrive.

from collections import deque

from datetime import datetime, timedelta

class SlidingWindowStats:
def __init__(self, window_seconds=300):
[Link] = timedelta(seconds=window_seconds)
[Link] = deque() # (timestamp, value)
[Link] = 0.0

def add(self, timestamp, value):

# Remove events outside the window
cutoff = timestamp - [Link]
while [Link] and [Link][0][0] < cutoff:
old_ts, old_val = [Link]()
[Link] -= old_val
[Link]((timestamp, value))
[Link] += value

@property
def count(self):
return len([Link])

@property
def mean(self):
return [Link] / len([Link]) if [Link] else 0.0

# Usage
stats = SlidingWindowStats(window_seconds=300)

for ts, val in event_stream:

[Link](ts, val)
print(f"Count: {[Link]}, Mean: {[Link]:.2f}")

# Time complexity: O(1) amortized per event (each event added/removed once)
# Space: O(events in window)

# Production version uses Flink's sliding window operator

# Or Spark Structured Streaming with window() function
# Or Snowflake's streams + tasks with rolling computation

# Follow-up: what if you need median, percentiles?

# - Naive: sort window each time (slow)
# - Better: maintain two heaps (min-heap for upper half, max-heap for lower)
# - Best for percentiles: use approximate (t-digest, GK summary)

Python + Coding Deep Dive Page 36

 Section B: Algorithm Pattern Problems (11-20)
Problem 11: Two-sum
Given: array of integers and target. Return: indices of two numbers summing to target. Classic, but
pattern shows up in joins, lookups, dedup.

# O(N) using hash map

def two_sum(nums, target):
seen = {} # value -> index
for i, num in enumerate(nums):
complement = target - num
if complement in seen:
return [seen[complement], i]
seen[num] = i
return None

# Pattern: this is what a hash join does in databases

# Build hash on smaller side; probe with larger side
# O(N + M) instead of O(N * M) for nested loop

Problem 12: Sliding window maximum

Given: array and window size K. Return: max in each sliding window of size K.

from collections import deque

def sliding_window_max(nums, k):

result = []
dq = deque() # stores indices; values in decreasing order

for i, num in enumerate(nums):

# Remove indices outside window
while dq and dq[0] < i - k + 1:
[Link]()
# Remove smaller values from back (they'll never be max)
while dq and nums[dq[-1]] < num:
[Link]()
[Link](i)
# First k-1 iterations don't have a full window
if i >= k - 1:
[Link](nums[dq[0]])
return result

# Time: O(N) amortized; each element added/removed at most once

# Space: O(K) for the deque

# Variant: sliding window MIN - same approach, opposite comparison

# Variant: streaming version - same data structure, no array index

Problem 13: Group anagrams

Python + Coding Deep Dive Page 37

 # Group strings by anagram class
from collections import defaultdict

def group_anagrams(strs):
groups = defaultdict(list)
for s in strs:
key = ''.join(sorted(s)) # canonical form
groups[key].append(s)
return list([Link]())

# Alternative key: char count tuple (faster for long strings)

def group_anagrams_v2(strs):
groups = defaultdict(list)
for s in strs:
count = [0] * 26
for c in s:
count[ord(c) - ord('a')] += 1
groups[tuple(count)].append(s)
return list([Link]())

# Time: O(N * M log M) for v1 where M = string length

# Time: O(N * M) for v2
# Pattern: canonical form for grouping - shows up in dedup, schema matching

Python + Coding Deep Dive Page 38

 Problem 14: Merge intervals

def merge_intervals(intervals):
if not intervals:
return []
[Link](key=lambda x: x[0])
merged = [intervals[0]]
for start, end in intervals[1:]:
last_start, last_end = merged[-1]
if start <= last_end:
merged[-1] = (last_start, max(last_end, end))
else:
[Link]((start, end))
return merged

# Pattern: range arithmetic. Same logic for:

# - Calendar conflict detection
# - Time-series gap detection
# - Lock conflict in concurrent systems
# - Sensor data merging

Problem 15: LRU Cache

from collections import OrderedDict

class LRUCache:
def __init__(self, capacity):
[Link] = capacity
[Link] = OrderedDict()

def get(self, key):

if key not in [Link]:
return -1
[Link].move_to_end(key)
return [Link][key]

def put(self, key, value):

if key in [Link]:
[Link].move_to_end(key)
[Link][key] = value
if len([Link]) > [Link]:
[Link](last=False)

# Time: O(1) for both get and put

# This is the pattern in: page cache, query result cache,
# warehouse local SSD cache, web browser cache

# Without OrderedDict (interviewer may ask):

# Use hash map + doubly linked list manually
# - Hash map gives O(1) lookup
# - Linked list gives O(1) reorder
# Both together = O(1) for all operations

Problem 16: BFS on graph - shortest path

Python + Coding Deep Dive Page 39

 from collections import deque

def shortest_path(graph, start, end):

if start == end:
return [start]
queue = deque([(start, [start])])
visited = {start}
while queue:
node, path = [Link]()
for neighbor in [Link](node, []):
if neighbor == end:
return path + [neighbor]
if neighbor not in visited:
[Link](neighbor)
[Link]((neighbor, path + [neighbor]))
return None # no path

# Time: O(V + E)
# Pattern: shortest path in unweighted graph
# Examples: data lineage traversal, dependency resolution in DAGs,
# downstream impact of a column change

Python + Coding Deep Dive Page 40

 Problem 17: Topological sort (DAG ordering)
Use case: ordering tasks in a pipeline, resolving dependencies, computing build order.

from collections import defaultdict, deque

def topological_sort(num_nodes, edges):

# edges: list of (from, to) representing dependencies
graph = defaultdict(list)
in_degree = [0] * num_nodes
for u, v in edges:
graph[u].append(v)
in_degree[v] += 1

# Start with nodes that have no dependencies

queue = deque([i for i in range(num_nodes) if in_degree[i] == 0])
result = []

while queue:
node = [Link]()
[Link](node)
for neighbor in graph[node]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
[Link](neighbor)

if len(result) != num_nodes:
raise ValueError("Cycle detected")
return result

# Time: O(V + E)
# This is the algorithm Airflow uses to order DAG task execution
# Also: dbt model execution order; Iceberg manifest dependency resolution

Problem 18: Median of streaming data

Python + Coding Deep Dive Page 41

 import heapq

class MedianFinder:
def __init__(self):
# Max-heap for lower half (negate values since heapq is min-heap)
[Link] = []
# Min-heap for upper half
[Link] = []

def add(self, num):

# Push to lower (max-heap)
[Link]([Link], -num)
# Move max of lower to upper
[Link]([Link], -[Link]([Link]))
# Rebalance if needed (lower should have equal or 1 more)
if len([Link]) > len([Link]):
[Link]([Link], -[Link]([Link]))

def median(self):
if len([Link]) > len([Link]):
return -[Link][0]
return (-[Link][0] + [Link][0]) / 2

# Time: O(log N) per add; O(1) per median query

# Space: O(N)
# Pattern: two heaps for streaming order statistics
# Use case: real-time p50 latency monitoring; quantile dashboards

Problem 19: Word frequency in huge file

Python + Coding Deep Dive Page 42

 # Memory-bounded word count for files bigger than RAM
from collections import Counter

def word_count_bounded(file_path, top_k=100):

# If file fits in memory:
with open(file_path) as f:
counter = Counter(word for line in f for word in [Link]())
return counter.most_common(top_k)

# If file is HUGE (terabytes):

# 1. Map: chunks of file -> partial Counter
# 2. Reduce: merge Counters
# 3. This is the MapReduce / Spark wordcount pattern

def parallel_word_count(file_paths, top_k=100):

from multiprocessing import Pool

def count_chunk(path):
counter = Counter()
with open(path) as f:
for line in f:
[Link]([Link]())
return counter

with Pool() as pool:

partial_counters = [Link](count_chunk, file_paths)

# Merge
total = Counter()
for c in partial_counters:
[Link](c)
return total.most_common(top_k)

# Spark version (one-liner):

# [Link]('[Link]').[Link](lambda x: [Link]())
# .map(lambda w: (w, 1)).reduceByKey(lambda a, b: a+b)
# .top(100, key=lambda x: x[1])

Python + Coding Deep Dive Page 43

 Problem 20: Detect cycle in directed graph

from collections import defaultdict

def has_cycle(num_nodes, edges):

graph = defaultdict(list)
for u, v in edges:
graph[u].append(v)

WHITE, GRAY, BLACK = 0, 1, 2

color = [WHITE] * num_nodes

def dfs(node):
if color[node] == GRAY: # found cycle (visited in current path)
return True
if color[node] == BLACK: # already processed
return False
color[node] = GRAY
for neighbor in graph[node]:
if dfs(neighbor):
return True
color[node] = BLACK
return False

return any(dfs(i) for i in range(num_nodes) if color[i] == WHITE)

# Time: O(V + E)
# Use case: detect cycles in dbt model dependencies, Airflow DAG validation,
# data lineage analysis - any DAG that should be acyclic

Python + Coding Deep Dive Page 44

 Section C: PySpark-Specific Problems (21-30)
Problem 21: Find users who bought product A and B (not OR)

# Bad: filter for either, no constraint

[Link](([Link]('product') == 'A') | ([Link]('product') == 'B')) # wrong

# Better: pivot pattern

from [Link] import functions as F

def users_with_both_products(purchases):
return (purchases
.groupBy('user_id')
.agg(
[Link]([Link]([Link]('product') == 'A', 1).otherwise(0)).alias('has_A'),
[Link]([Link]([Link]('product') == 'B', 1).otherwise(0)).alias('has_B'),
)
.filter(([Link]('has_A') == 1) & ([Link]('has_B') == 1))
.select('user_id')
)

# Alternative: count distinct products in filtered set

def users_with_both_v2(purchases):
return (purchases
.filter([Link]('product').isin(['A', 'B']))
.groupBy('user_id')
.agg([Link]('product').alias('distinct_products'))
.filter([Link]('distinct_products') == 2)
.select('user_id')
)

Problem 22: Compute conversion rate per funnel step

Python + Coding Deep Dive Page 45

 # Given events: ('user_id', 'event_type', 'event_time')
# Funnel: view -> add_to_cart -> checkout -> purchase
# Return: at each step, % of users who continued

from [Link] import functions as F

def funnel_conversion(events):
funnel_steps = ['view', 'add_to_cart', 'checkout', 'purchase']

# For each user, find max funnel step reached

step_rank = F.create_map(*sum(
[[[Link](s), [Link](i)] for i, s in enumerate(funnel_steps)], []
))

user_max_step = (events
.filter([Link]('event_type').isin(funnel_steps))
.withColumn('step_idx', step_rank[[Link]('event_type')])
.groupBy('user_id')
.agg([Link]('step_idx').alias('max_step'))
)

# Count users at each step or higher

step_counts = []
for i, step in enumerate(funnel_steps):
count = user_max_step.filter([Link]('max_step') >= i).count()
step_counts.append((step, count))

# Compute conversion %
base = step_counts[0][1]
return [(step, count, count / base * 100) for step, count in step_counts]

# Production pattern: pre-compute as a daily dbt model

# Used at: every e-commerce / SaaS company; classic analytics question

Python + Coding Deep Dive Page 46

 Problem 23: Handle skewed join

# Problem: joining 'events' (1B rows) on 'user_id' but a few power users
# have 10% of all events. Single Spark task gets 100M rows = OOM.

# Diagnosis (run first):

[Link]('user_id').count().orderBy([Link]('count').desc()).show(10)
# If top key counts are 100x median, you have skew.

# Fix 1: Enable AQE skew join (Spark 3.0+; usually default)

[Link]('[Link]', 'true')
[Link]('[Link]', 'true')

# Fix 2: Salt the hot keys manually (for older Spark or extreme skew)
NUM_SALTS = 100

# Salt the large side

events_salted = [Link](
'salted_user_id',
[Link]([Link]('user_id'), [Link]('_'),
([Link]() * NUM_SALTS).cast('int'))
)

# Replicate the small side (each row N times with all salt values)
salts_array = [Link]([[Link](i) for i in range(NUM_SALTS)])
users_salted = (users
.withColumn('salt', [Link](salts_array))
.withColumn('salted_user_id',
[Link]([Link]('user_id'), [Link]('_'), [Link]('salt')))
)

# Now join on salted key - load distributed

joined = events_salted.join(users_salted, 'salted_user_id')
# Drop salted_user_id; keep results
result = [Link]('salted_user_id', 'salt')

Problem 24: Read changing schema CSVs

Python + Coding Deep Dive Page 47

 # Files in S3 have evolved over time:
# 2025/01/*.csv has columns: id, name, amount
# 2026/01/*.csv has columns: id, name, amount, category

# Strategy 1: Schema merge (works for additive changes)

df = ([Link]
.option('mergeSchema', 'true')
.option('header', 'true')
.csv('s3://bucket/data/'))
# Missing columns = NULL for older rows

# Strategy 2: Read with explicit schema, ignore extras

from [Link] import StructType, StructField, StringType, DoubleType
schema = StructType([
StructField('id', StringType()),
StructField('name', StringType()),
StructField('amount', DoubleType()),
StructField('category', StringType()), # NULL for old files
])
df = [Link](schema).csv('s3://bucket/data/', header=True)

# Strategy 3: Per-year processing

import os
old = [Link]('s3://bucket/data/2025/', header=True)
new = [Link]('s3://bucket/data/2026/', header=True)
old = [Link]('category', [Link](None).cast('string'))
combined = [Link](new, allowMissingColumns=True)

# For production: convert all to Iceberg/Delta and use schema evolution

Problem 25: Detect duplicate orders within 5 seconds

# Use case: idempotency / fraud detection

# Same user, same amount, within 5 seconds = likely duplicate

from [Link] import Window

import [Link] as F

def detect_duplicates(orders):
w = [Link]('user_id', 'amount').orderBy('order_time')

return (orders
.withColumn('prev_order_time', [Link]('order_time').over(w))
.withColumn('seconds_since_prev',
F.unix_timestamp('order_time') - F.unix_timestamp('prev_order_time'))
.filter([Link]('seconds_since_prev').isNotNull() &
([Link]('seconds_since_prev') <= 5))
)

# Returns the SECOND (and later) duplicates

# Pair with the action: alert, refund, mark as suspicious

# Variant: detect by user + amount + IP within window

# Extend partition: w = [Link]('user_id', 'amount', 'client_ip')

Python + Coding Deep Dive Page 48

 Problem 26: Compute Median per group

# Spark's approxQuantile is fast but only at DataFrame level

# For per-group quantiles, use percentile_approx

import [Link] as F

result = ([Link]('region').agg(
F.percentile_approx('latency_ms', 0.5).alias('median'),
F.percentile_approx('latency_ms', 0.95).alias('p95'),
F.percentile_approx('latency_ms', 0.99).alias('p99'),
F.percentile_approx('latency_ms', [0.5, 0.95, 0.99]).alias('percentiles'),
))

# Exact median is expensive - requires sorting

# percentile_approx uses t-digest internally; ~1% accurate

Problem 27: Slow Changing Dimension (SCD Type 2) update

Python + Coding Deep Dive Page 49

 # Given: dim_customer with effective_from, effective_to, is_current
# New batch of customer updates
# Update dim_customer to preserve history

from [Link] import functions as F

def apply_scd2(current_dim, new_records, key_col='customer_id'):

# 1. Find records that changed (exist in both, but differ)
changed = (new_records.alias('n')
.join(current_dim.filter([Link]('is_current')).alias('c'),
on=key_col, how='inner')
.where([Link]('[Link]') != [Link]('[Link]')) # detect change
.select('n.*')
)

# 2. Mark old records as expired

update_old = (current_dim
.filter([Link]('is_current') &
[Link](key_col).isin([r[key_col] for r in [Link]()]))
.withColumn('effective_to', F.current_timestamp())
.withColumn('is_current', [Link](False))
)

# 3. Insert new versions

insert_new = (changed
.withColumn('effective_from', F.current_timestamp())
.withColumn('effective_to', [Link](None).cast('timestamp'))
.withColumn('is_current', [Link](True))
)

# 4. Also insert brand-new keys

new_keys = (new_records.alias('n')
.join(current_dim.alias('c'), on=key_col, how='left_anti')
.withColumn('effective_from', F.current_timestamp())
.withColumn('effective_to', [Link](None).cast('timestamp'))
.withColumn('is_current', [Link](True))
)

# Union everything
return (current_dim.subtract(update_old.select(current_dim.columns))
.union(update_old.select(current_dim.columns))
.union(insert_new.select(current_dim.columns))
.union(new_keys.select(current_dim.columns))
)

# In production: use Delta Lake or Iceberg MERGE INTO for atomic SCD2

Problem 28: Pre-aggregation for cube

Python + Coding Deep Dive Page 50

 # Compute aggregates across multiple dimension combinations efficiently
# Sales by: country, country+product, country+product+date, etc.

from [Link] import functions as F

# Using cube() - all combinations of dimensions

result = ([Link]('country', 'product', 'date')
.agg(
[Link]('revenue').alias('total_revenue'),
[Link]('*').alias('order_count'),
)
)
# Result includes rows with NULL for "all values" of a dimension

# Using rollup() - hierarchical aggregation

result = ([Link]('country', 'product', 'date')
.agg([Link]('revenue').alias('total_revenue'))
)
# rollup is a subset of cube; useful for hierarchies (geography rollup)

# Using grouping_sets - explicit subset of cube combinations

from [Link] import grouping_id
result = [Link]('country', 'product').agg(
[Link]('revenue'),
grouping_id().alias('level') # tells you which dims are aggregated
)

Python + Coding Deep Dive Page 51

 Problem 29: Reading and writing partitioned Iceberg

# Write Iceberg table with partitioning

([Link]('[Link]')
.partitionedBy([Link]('event_time'), [Link](16, 'user_id'))
.create()
)

# Subsequent writes (Iceberg handles partition evolution)

[Link]('[Link]').append()

# Read with predicate that benefits from partitioning

result = ([Link]('[Link]')
.filter([Link]('event_time') >= '2026-05-01')
.filter([Link]('user_id') == 'specific_user')
)
# Iceberg prunes by days partition AND user_id bucket
# Sub-second on multi-TB table

# Time travel
old_data = ([Link]
.option('snapshot-id', '12345')
.table('[Link]')
)

# Compaction (call procedure)

[Link]('''
CALL [Link].rewrite_data_files(
table => '[Link]',
options => map('target-file-size-bytes', '536870912')
)
''')

Problem 30: End-to-end ingest pipeline

Given: design a PySpark job that ingests daily CSV from S3, dedupes, joins reference data, writes to
Iceberg, with error handling and validation.

Python + Coding Deep Dive Page 52

 from [Link] import SparkSession, functions as F
import logging

def daily_ingest_pipeline(spark, date_str):

logger = [Link](__name__)

# 1. Read raw data

raw_path = f's3://bucket/raw/orders/dt={date_str}/'
try:
raw_df = [Link](raw_path, header=True, inferSchema=True)
except Exception as e:
[Link](f"Failed to read {raw_path}: {e}")
raise

raw_count = raw_df.count()
[Link](f"Read {raw_count} raw rows for {date_str}")

# 2. Validate (fail fast if bad data)

if raw_count == 0:
raise ValueError(f"No data for {date_str}")

# 3. Clean + deduplicate
cleaned = (raw_df
.filter([Link]('order_id').isNotNull())
.filter([Link]('amount') > 0)
.dropDuplicates(['order_id'])
)

# 4. Join reference data (customers)

customers = [Link]('[Link]')
enriched = [Link]([Link](customers), 'customer_id', 'left')

# 5. Add metadata
final = (enriched
.withColumn('ingested_at', F.current_timestamp())
.withColumn('source_file_date', [Link](date_str))
)

# 6. Quality check before write

null_check = [Link]([Link]('customer_id').isNull()).count()
if null_check > raw_count * 0.05:
raise ValueError(f"More than 5% null customer_id: {null_check}")

# 7. Write to Iceberg
([Link]('[Link].fact_orders')
.partitionedBy([Link]('order_date'))
.append()
)

[Link](f"Wrote {[Link]()} rows to fact_orders")

return [Link]()

# In Airflow:
# t1 = PythonOperator(task_id='ingest_orders',
# python_callable=daily_ingest_pipeline,
# op_kwargs={'date_str': '{{ ds }}'})

Python + Coding Deep Dive Page 53

 PART 5
Testing Data Pipelines

5.1 Why testing data pipelines is different

Testing pipelines is harder than testing application code because: data shape changes; tests need
representative data; side effects (writes to warehouse) are harder to isolate; full end-to-end is slow.
Strategies need to adapt.

The 4 layers of pipeline tests

Layer What it tests Speed When run

Unit tests Pure functions in isolation Fast (ms) Every commit

Integration tests Component interactions (e.g., Spark Medium (sec) Every commit
transformations)

End-to-end tests Full pipeline against sample data Slow (min) Before release

Data quality tests Output validation in production Always After every run

5.2 pytest patterns for DE

Python + Coding Deep Dive Page 54

 # tests/test_transformations.py
import pytest
from datetime import datetime
from [Link] import compute_discount, sessionize_events

# Basic test
def test_compute_discount_gold():
assert compute_discount(100, 'gold') == 20

# Parametrize - test multiple cases concisely

@[Link]("amount,tier,expected", [
(100, 'gold', 20),
(100, 'silver', 10),
(100, 'bronze', 0),
(0, 'gold', 0),
(-50, 'gold', 0), # edge case: negative
])
def test_compute_discount_cases(amount, tier, expected):
assert compute_discount(amount, tier) == expected

# Fixtures - reusable setup

@[Link]
def sample_events():
return [
{'user_id': 'u1', 'event_time': '2026-05-01T10:00:00', 'type': 'view'},
{'user_id': 'u1', 'event_time': '2026-05-01T10:05:00', 'type': 'click'},
{'user_id': 'u1', 'event_time': '2026-05-01T11:00:00', 'type': 'view'},
# Gap > 30min, so new session
]

def test_sessionize_creates_two_sessions(sample_events):
sessions = sessionize_events(sample_events, gap_minutes=30)
assert len(sessions) == 2

# Fixture with finalizer (cleanup)

@[Link]
def temp_database():
db = create_test_db()
yield db
[Link]()

def test_with_db(temp_database):
temp_database.insert({'id': 1})
assert temp_database.count() == 1
# Cleanup runs automatically after test

# Mocking
from [Link] import patch, MagicMock

def test_api_call_with_retry():
with patch('[Link]') as mock_get:
mock_get.side_effect = [
Exception('Timeout'),
MagicMock(status_code=200, json=lambda: {'ok': True})
]
result = fetch_with_retry('[Link]
assert result == {'ok': True}
assert mock_get.call_count == 2

Python + Coding Deep Dive Page 55

 5.3 Testing PySpark code
# tests/test_spark_transformations.py
import pytest
from [Link] import SparkSession
from src.spark_transformations import deduplicate, top_n_per_group

# Spark session fixture - shared across tests in module

@[Link](scope='module')
def spark():
return ([Link]
.master('local[*]')
.appName('tests')
.config('[Link]', '2') # small for tests
.getOrCreate())

def test_deduplicate_removes_duplicates(spark):
input_data = [
('u1', '2026-05-01', 100),
('u1', '2026-05-02', 100), # newer, should win
('u2', '2026-05-01', 50),
]
df = [Link](input_data, ['user_id', 'date', 'amount'])

result = deduplicate(df, key='user_id', order_by='date')

rows = [Link]()

assert len(rows) == 2
u1_row = next(r for r in rows if r.user_id == 'u1')
assert u1_row.date == '2026-05-02' # latest

# Use chispa for nicer DataFrame assertions

# pip install chispa
from chispa.dataframe_comparer import assert_df_equality

def test_top_n_per_group(spark):
input_df = [Link]([
('u1', 100), ('u1', 200), ('u1', 50),
('u2', 300), ('u2', 150),
], ['user_id', 'amount'])

expected = [Link]([
('u1', 200), ('u1', 100),
('u2', 300), ('u2', 150),
], ['user_id', 'amount'])

result = top_n_per_group(input_df, partition='user_id', order='amount', n=2)

assert_df_equality(result, expected, ignore_row_order=True)

5.4 Production data quality tests

Different from unit/integration tests - these run AFTER each production pipeline run, validating actual
output.

Python + Coding Deep Dive Page 56

 # Using Great Expectations (popular framework)
import great_expectations as ge

def validate_fact_orders(df):
suite = ge.from_pandas([Link]()) # or use spark-native

# Schema expectations
suite.expect_table_columns_to_match_set([
'order_id', 'customer_id', 'amount', 'order_date', 'ingested_at'
])

# Row count expectations

suite.expect_table_row_count_to_be_between(min_value=1000, max_value=10000000)

# Null expectations
suite.expect_column_values_to_not_be_null('order_id')
suite.expect_column_values_to_not_be_null('customer_id')

# Uniqueness
suite.expect_column_values_to_be_unique('order_id')

# Value distribution
suite.expect_column_values_to_be_between('amount', min_value=0, max_value=100000)

# Run validation
result = [Link]()
if not [Link]:
raise DataQualityError(f"Validation failed: {result}")

# Alternative: dbt tests (declarative in YAML)

# models/fact_orders.yml
# tests:
# - unique: {column: order_id}
# - not_null: {column: customer_id}
# - relationships: {to: ref('dim_customer'), field: customer_id}

# Run via dbt test

# These integrate naturally if you're already in dbt

Python + Coding Deep Dive Page 57

 PART 6
Live Coding Round Tactics

6.1 The 45-minute structure

Most Tier A/S coding rounds are 45-60 minutes. Time discipline matters as much as correct code.

# Recommended time allocation (45-min round)

Minutes 0-5: CLARIFY

- Restate the problem in your own words
- Ask about input format, size, edge cases
- Ask about expected output format
- Ask about constraints (memory, language, time)
- "Should I optimize for readability or performance?"

Minutes 5-10: DESIGN

- Discuss approach BEFORE coding
- Mention naive solution first
- Then propose optimized approach
- Get interviewer's agreement on direction
- State complexity (time + space) upfront

Minutes 10-30: CODE

- Implement the agreed approach
- Talk through what you're doing
- Type carefully but don't over-perfect
- Use meaningful variable names
- Add docstring/comment if logic is non-obvious

Minutes 30-40: TEST

- Walk through with a small example
- Test edge cases: empty input, single element, duplicates,
nulls, negative numbers
- If interviewer pushes back, engage with their concern

Minutes 40-45: EXTEND

- Discuss how you'd handle 100x scale
- Discuss what you'd add for production
- Discuss alternative approaches
- This is where Staff signal is shown

6.2 Think out loud - the principles

Silence kills. If you go silent for 60 seconds, the interviewer doesn't know if you're thinking deeply, stuck,
or confused. They can't help you and can't score you positively.

Narrate your thinking:

Python + Coding Deep Dive Page 58

 # Good narration patterns:

"Let me think out loud about this..."

"My first instinct is to use a hash map because..."
"I'm considering two approaches: A and B. A is simpler but..."
"Before I code this, let me verify with an example..."
"I'm trying to decide if I need to handle the null case here..."
"Let me trace through this with [1, 2, 3] to make sure it works..."

# When stuck:
"I'm stuck. Let me think about what I do know..."
"The constraint that's making this hard is X. If I could relax it..."
"Let me try a different angle - what if I sort the data first?"

# When you make a mistake:

"Wait, that's not right. Let me reconsider..."
"Actually, I think my approach has a bug at edge case X..."

# These all demonstrate engagement, problem-solving, self-correction

# Even if you don't reach optimal solution, you've shown strong process

Python + Coding Deep Dive Page 59

 6.3 Handling the 'what if scale was 100x' followup
Almost every Tier A/S coding round ends with this. Preparing for it = differentiator.

# Common scale follow-ups and answers:

Q: "What if the input was 1TB?"

A: "Currently I'm using a hash map with O(N) memory. At 1TB I can't fit
everything. I'd partition by key and process each partition separately
(MapReduce style). For Python: process in chunks; for production
use Spark."

Q: "What if requests came at 100K QPS?"

A: "Current logic is single-machine. I'd shard by key across N machines
(consistent hashing). State per shard. For hot keys: salt to spread
load (covered with example)."

Q: "What if data could be unsorted and arrive late?"

A: "Buffer for a watermark period to handle late arrivals. Use streaming
framework like Flink with allowed_lateness. Trade-off: more memory
+ slight latency for correctness."

Q: "How would you run this in production?"

A: "Three things I'd add: (1) Tests - unit + integration + data quality
checks on output; (2) Observability - log key metrics, alert on
anomalies; (3) Idempotency - safe to re-run on same input."

Q: "What if you needed real-time?"

A: "Currently batch-oriented. For real-time: ingest via Kafka, process with
Flink stateful operators, write to KV store (Redis) for serving.
Trade memory + complexity for latency."

# Pattern: every follow-up has a recognizable answer template.

# Memorize the templates; adapt the specifics.

6.4 Recovery scripts

Scenario: You finished too quickly

Script

'I have a working solution but I want to check edge cases I might have missed.' [Walk through 3 edge
cases.] 'Also, let me consider what would change at 100x scale.' [Discuss.] 'Anything specific you'd
like me to dive deeper on?'

Scenario: You're 30 min in and not done

Script

'I notice we're running short on time. Let me describe what I'd do for the remaining pieces rather than
coding them in detail.' [Describe the rest at high level.] 'Would you like me to code any specific part of
that, or is the high level sufficient?'

Scenario: Solution doesn't work, time running out

Python + Coding Deep Dive Page 60

 Script

'I can see there's a bug here. Let me explain what I THINK is happening and how I'd debug it.' [Walk
through your debugging logic.] 'Given more time, I'd add print statements at X and verify the invariant
at Y.' Showing debugging process is valuable even without a working solution.

Scenario: Interviewer suggests different approach

Script

DON'T defensively defend your approach. DO: 'Interesting, let me think about that. The trade-offs
would be... I see the benefit is... The cost would be...' Engage genuinely, then either pivot or explain
why your approach still fits.

Python + Coding Deep Dive Page 61

 6.5 The 5 things interviewers actually score on
After dozens of debriefs, here's what really matters in the score:

1. Problem understanding (15%). Did you grasp what was being asked? Did you ask the right
clarifying questions?
2. Approach (25%). Did you propose a reasonable approach before coding? Did you discuss
trade-offs?
3. Code quality (20%). Is the code readable, modular, correctly handling edge cases?
4. Communication (20%). Did you think out loud? Engage with feedback? Make the interviewer's job
easy?
5. Production thinking (20%). Did you discuss scaling, testing, error handling, observability?
Perfect solution + bad communication = often rejected. Average solution + strong communication +
good discussion of trade-offs = often offered. Communication and process matter as much as code.

Killer detail

Most candidates focus 100% on the code. The most-improved candidates after practice spend equal
time on communication. Practice talking through problems out loud, even alone. Record yourself;
listen back. The growth happens here.

Python + Coding Deep Dive Page 62

 Summary + Practice Schedule
Self-check questions
# Question

1 Mutable default arguments - explain the gotcha and the fix

2 List comprehension vs generator - when use each

3 Write a retry decorator from memory

4 Pandas vs Polars - when use each

5 PyArrow's role in the data ecosystem

6 Why are PySpark UDFs slow? What's the alternative?

7 Broadcasting, repartitioning, coalescing - when use each

8 Reading a Spark explain plan - what to look for

9 Solve rolling 7-day avg per customer (Problem 1)

10 Solve sessionization (Problem 2)

11 Solve top-K streaming (Problem 3)

12 Solve SCD2 update (Problem 27)

13 Solve skewed join (Problem 23)

14 45-min coding round time allocation - recite the structure

15 'What if scale was 100x' - have a template answer ready

Practice schedule (2 weeks)

Week 1 - Coverage:

Day 1: Read Parts 1 + 2. Note any concepts you've never used in production.
Day 2: Read Part 3 (PySpark). Set up local Spark environment if you don't have one.
Day 3: Read Part 5 (testing). Write 5 tests for Problem 1 yourself.
Day 4-5: Work through Problems 1-10. Timer ON. Solution AFTER attempt.
Day 6-7: Work through Problems 11-20.
Week 2 - Depth:

Day 8-9: Work through Problems 21-30 (PySpark).

Day 10: Re-solve the 5 problems you found hardest (without looking at solutions).
Day 11: Mock coding interview. Record yourself.
Day 12: Listen to recording. Note communication gaps. Read Part 6 (tactics).
Day 13: Second mock. Apply tactical improvements.
Day 14: Light review + rest.

Python + Coding Deep Dive Page 63

 Final thought
Python coding round depth is closed only through reps. This doc gives you the framework, the patterns,
and the 30 highest-leverage problems. The transformation from 'reads about coding' to 'can code under
pressure' happens when you struggle with each problem yourself, fail at edge cases, and iterate.

If you do the 30 problems with a timer over 2 weeks, you'll be in the top 20% of DE candidates on the
coding round. Combined with strong system design (Playbook) and behavioral (Weeks 3-4 + 9-10), you
have the complete interview toolkit.

Python + Coding Deep Dive Page 64