Python Interview Guide
For AWS Data Engineers — 3 Years Experience
Theory + Hands-on Coding Questions with Answers
1. Python Core Concepts (Theory)
2. Data Structures & Algorithms (Coding)
3. File Handling & Data Formats
4. Pandas / PySpark
5. boto3 & AWS-specific Python
6. Error Handling, Logging & Testing
7. Performance & Concurrency
1. Python Core Concepts (Theory)
Q1. What is the difference between mutable and immutable types in Python? Why does it
matter in a data pipeline?
Answer: Immutable objects (int, str, tuple, frozenset) cannot be changed after creation — any 'modification'
creates a new object. Mutable objects (list, dict, set) can be changed in place. This matters in pipelines
because passing a mutable object (e.g., a list of records) into a function and modifying it can cause
unintended side effects across stages of a pipeline if multiple functions share the same reference. It also
affects hashability — only immutable objects can be dict keys or set members.
Q2. Differentiate list, tuple, set, and dict — when would you use each in an ETL context?
Answer: List: ordered, mutable — good for an ordered batch of records being processed sequentially.
Tuple: ordered, immutable — good for fixed-structure rows (e.g., a row coming from a DB cursor) since it
can't be altered accidentally. Set: unordered, unique elements — good for deduplication or fast membership
checks (O(1) lookup) like checking processed IDs. Dict: key-value, fast lookups — good for lookups/joins in
memory, e.g., mapping IDs to reference data during enrichment.
Q3. Why prefer generators over lists when processing large files (e.g., from S3)?
Answer: Generators produce items lazily, one at a time, using O(1) memory regardless of dataset size,
instead of loading the entire dataset into memory like a list does. For large files this avoids MemoryError and
reduces time-to-first-result since downstream processing can start before the whole file is read.
def read_large_file(path):
with open(path) as f:
for line in f:
yield [Link]()
# Memory-safe even for a multi-GB file
for record in read_large_file("big_data.csv"):
process(record)
Q4. Explain *args, **kwargs, and write a decorator that retries a function on failure.
Answer: *args collects extra positional arguments into a tuple; **kwargs collects extra keyword arguments
into a dict. They let you write generic wrappers (decorators) that work with any function signature.
import time
import functools
def retry(times=3, delay=2):
def decorator(func):
@[Link](func)
def wrapper(*args, **kwargs):
last_exc = None
for attempt in range(1, times + 1):
try:
return func(*args, **kwargs)
except Exception as e:
last_exc = e
print(f"Attempt {attempt} failed: {e}")
[Link](delay)
raise last_exc
return wrapper
return decorator
@retry(times=3, delay=1)
def fetch_s3_object(bucket, key):
# boto3 call that might fail transiently
...
Q5. What is the GIL? How does it influence the choice between multithreading and
multiprocessing in Python data pipelines?
Answer: The Global Interpreter Lock (GIL) allows only one thread to execute Python bytecode at a time in
CPython. Threading is therefore not useful for CPU-bound work (e.g., heavy transformations, parsing) since
threads compete for the GIL — use multiprocessing instead, which uses separate processes with their own
interpreter and memory, bypassing the GIL. Threading still works well for I/O-bound work (S3 downloads,
API calls, DB queries) since the GIL is released during I/O waits.
Q6. What are context managers? Write a custom one for managing a resource like a DB or S3
connection.
Answer: Context managers (used with the 'with' statement) guarantee setup/teardown code runs, even if an
exception occurs — implemented via __enter__/__exit__ methods or the @contextmanager decorator.
from contextlib import contextmanager
import boto3
@contextmanager
def s3_client():
client = [Link]("s3")
try:
yield client
finally:
[Link]() # cleanup guaranteed even on exception
with s3_client() as s3:
s3.upload_file("[Link]", "my-bucket", "data/[Link]")
Q7. Difference between shallow copy and deep copy?
Answer: A shallow copy ([Link] / list(x) / dict(x)) creates a new outer object but nested objects are still
shared references. A deep copy ([Link]) recursively copies nested objects too, so changes to
nested data in the copy don't affect the original. This matters when duplicating config dicts or nested JSON
records before mutating them in a pipeline step.
Q8. How do you design custom exceptions for a pipeline, and why not just use generic
exceptions?
Answer: Custom exceptions (subclassing Exception) let calling code distinguish failure types and react
appropriately (e.g., retry on a transient error, fail-fast on a data-validation error) instead of catching a broad
Exception and guessing.
class DataValidationError(Exception):
pass
class TransientSourceError(Exception):
pass
def load_record(record):
if [Link]("id") is None:
raise DataValidationError(f"Missing id in record: {record}")
2. Data Structures & Algorithms (Coding)
Q9. Remove duplicates from a list while preserving order.
def dedupe_preserve_order(items):
seen = set()
result = []
for item in items:
if item not in seen:
[Link](item)
[Link](item)
return result
# O(n) time, O(n) space
print(dedupe_preserve_order([3, 1, 3, 2, 1, 4])) # [3, 1, 2, 4]
Q10. Find the second-largest element in a list without using sort().
def second_largest(nums):
first = second = float('-inf')
for n in nums:
if n > first:
first, second = n, first
elif first > n > second:
second = n
return second
print(second_largest([10, 5, 20, 8, 20])) # 10
Q11. Group a list of dicts by a key (e.g., sales records by region).
from collections import defaultdict
def group_by(records, key):
grouped = defaultdict(list)
for r in records:
grouped[r[key]].append(r)
return grouped
sales = [{"region": "US", "amt": 100}, {"region": "EU", "amt": 50}, {"region": "US", "amt": 75}]
print(group_by(sales, "region"))
Q12. Flatten a nested dict (common when parsing JSON from an API).
def flatten(d, parent_key="", sep="."):
items = {}
for k, v in [Link]():
new_key = f"{parent_key}{sep}{k}" if parent_key else k
if isinstance(v, dict):
[Link](flatten(v, new_key, sep))
else:
items[new_key] = v
return items
nested = {"user": {"id": 1, "address": {"city": "NY"}}}
print(flatten(nested)) # {'[Link]': 1, '[Link]': 'NY'}
Q13. Count word frequency from a very large text file using streaming (constant memory for
the read).
from collections import Counter
def word_frequency(path):
counter = Counter()
with open(path) as f:
for line in f:
[Link]([Link]())
return counter
top10 = word_frequency("big_log.txt").most_common(10)
Q14. Deduplicate rows (list of dicts) based on a key, keeping the latest by timestamp.
def dedupe_latest(rows, key, ts_field):
latest = {}
for row in rows:
existing = [Link](row[key])
if existing is None or row[ts_field] > existing[ts_field]:
latest[row[key]] = row
return list([Link]())
rows = [
{"id": 1, "val": "a", "ts": 100},
{"id": 1, "val": "b", "ts": 200},
{"id": 2, "val": "c", "ts": 150},
]
print(dedupe_latest(rows, "id", "ts"))
Q15. Merge two sorted lists/iterators efficiently (relevant to merging sorted partitions).
import heapq
def merge_sorted(*iterables):
return [Link](*iterables) # O(n log k), lazy/streaming
a = [1, 4, 7]
b = [2, 3, 9]
print(list(merge_sorted(a, b))) # [1, 2, 3, 4, 7, 9]
3. File Handling & Data Formats
Q16. How do you read a large CSV file without loading it fully into memory?
Answer: Use pandas with chunksize (returns an iterator of DataFrames) or plain Python's csv module which
reads row by row, rather than pandas.read_csv() with no chunksize, which loads everything at once.
import pandas as pd
total = 0
for chunk in pd.read_csv("huge_file.csv", chunksize=100_000):
total += chunk["amount"].sum()
print(total)
Q17. CSV vs Parquet — what are the tradeoffs, and which would you choose for a data lake?
Answer: CSV: human-readable, simple, but row-based, no schema/types embedded, slow for analytics, no
compression by default, and reading requires scanning all columns. Parquet: columnar, compressed, stores
schema and types, supports predicate/column pushdown so analytical queries (e.g., Athena, Spark) read
only needed columns — much faster and cheaper for large-scale analytics. For a data lake feeding
Athena/Glue/Redshift Spectrum, Parquet (often partitioned) is generally preferred; CSV is fine for small files
or raw landing zones meant for human inspection.
Q18. Write code to read a JSON file and flatten nested fields into a flat table structure.
import json
import pandas as pd
def flatten(d, parent_key="", sep="_"):
items = {}
for k, v in [Link]():
new_key = f"{parent_key}{sep}{k}" if parent_key else k
if isinstance(v, dict):
[Link](flatten(v, new_key, sep))
else:
items[new_key] = v
return items
with open("[Link]") as f:
records = [Link](f)
flat_records = [flatten(r) for r in records]
df = [Link](flat_records)
Q19. How would you process a 10GB file on a machine with only 4GB RAM using Python?
Answer: Avoid loading the whole file into memory: stream it line-by-line or in chunks (pandas chunksize, or
plain file iteration), process and aggregate incrementally, and write/flush results out periodically instead of
accumulating everything in memory. If the file is on S3, you can also stream it directly using boto3's
StreamingBody, or split it into smaller partitions in S3 and process them independently (e.g., in parallel
workers or a Spark/Glue job) rather than handling the whole thing in one Python process.
4. Pandas / PySpark
Q20. Difference between apply(), map(), and vectorized operations in pandas — what are the
performance implications?
Answer: map() applies a function element-wise on a Series only. apply() applies a function row-wise or
column-wise on a Series/DataFrame and can be more flexible but is implemented as a Python-level loop
under the hood — slow on large data. Vectorized operations (e.g., df['a'] + df['b'], [Link]) use compiled C
code under pandas/NumPy and operate on whole arrays at once — typically 10-100x faster. Prefer
vectorized operations whenever possible; reserve apply() for logic that genuinely can't be vectorized.
Q21. How do you handle missing values in a DataFrame? Show a few strategies.
import pandas as pd
import numpy as np
df = [Link]({"a": [1, [Link], 3], "b": [[Link], 2, 3]})
[Link]().sum() # detect missing values per column
[Link]() # drop rows with any NaN
[Link](0) # fill with a constant
df["a"].fillna(df["a"].mean()) # fill with column mean
[Link](method="ffill") # forward-fill (time series)
Q22. Deduplicate rows in pandas based on a subset of columns, keeping the latest by
timestamp.
df = df.sort_values("updated_at").drop_duplicates(subset=["id"], keep="last")
Q23. Explain groupby().agg() with a multi-column aggregation example.
result = [Link]("region").agg(
total_sales=("amount", "sum"),
avg_sales=("amount", "mean"),
order_count=("order_id", "count")
).reset_index()
Q24. PySpark: difference between transformation and action; what is lazy evaluation?
Answer: Transformations (filter, select, map, groupBy, join) define a new DataFrame/RDD but don't execute
immediately — Spark builds a logical execution plan (DAG). Actions (collect, count, write, show) trigger
actual execution of that plan. Lazy evaluation means Spark waits until an action is called, allowing it to
optimize the whole chain of transformations (predicate pushdown, combining steps) before running, rather
than executing each step eagerly.
Q25. How do you avoid driver out-of-memory (OOM) errors in PySpark?
Answer: Avoid collect()/toPandas() on large DataFrames — they pull all data to the driver. Instead, use
actions like write(), count(), or take(n) for small samples; let executors do heavy aggregation; repartition data
appropriately; and use broadcast joins only for genuinely small lookup tables (the broadcast also lands on
every executor, not just the driver, so size it carefully).
5. boto3 & AWS-specific Python
Q26. Write code to upload and download a file to/from S3 using boto3.
import boto3
s3 = [Link]("s3")
# Upload
s3.upload_file("local_file.csv", "my-bucket", "raw/local_file.csv")
# Download
s3.download_file("my-bucket", "raw/local_file.csv", "[Link]")
Q27. How do you read a CSV directly from S3 into a pandas DataFrame without downloading it
locally first?
import pandas as pd
import boto3
s3 = [Link]("s3")
obj = s3.get_object(Bucket="my-bucket", Key="raw/[Link]")
df = pd.read_csv(obj["Body"]) # Body is a StreamingBody, pandas reads it directly
# Alternative (simplest, needs s3fs installed):
df = pd.read_csv("s3://my-bucket/raw/[Link]")
Q28. Write a Lambda function (Python) triggered by an S3 event that processes a file and
writes results to another bucket.
import boto3
import pandas as pd
import io
s3 = [Link]("s3")
def lambda_handler(event, context):
record = event["Records"][0]
src_bucket = record["s3"]["bucket"]["name"]
src_key = record["s3"]["object"]["key"]
obj = s3.get_object(Bucket=src_bucket, Key=src_key)
df = pd.read_csv(obj["Body"])
df["processed"] = True # example transformation
buffer = [Link]()
df.to_csv(buffer, index=False)
s3.put_object(
Bucket="processed-bucket",
Key=f"processed/{src_key}",
Body=[Link]()
)
return {"status": "success", "rows": len(df)}
Q29. How do you handle pagination when listing thousands of S3 objects with boto3?
import boto3
s3 = [Link]("s3")
paginator = s3.get_paginator("list_objects_v2")
keys = []
for page in [Link](Bucket="my-bucket", Prefix="raw/"):
for obj in [Link]("Contents", []):
[Link](obj["Key"])
print(f"Found {len(keys)} objects")
Q30. How would you implement retry with exponential backoff for a boto3 API call?
Answer: boto3/botocore already retries throttling errors by default, but you can tune it explicitly via Config,
or wrap calls yourself for custom logic.
import boto3
from [Link] import Config
config = Config(
retries={"max_attempts": 5, "mode": "adaptive"} # exponential backoff built in
)
s3 = [Link]("s3", config=config)
Q31. Give an example of using boto3 to interact with AWS Glue or DynamoDB.
import boto3
# Start a Glue job
glue = [Link]("glue")
response = glue.start_job_run(JobName="my-etl-job")
# Write an item to DynamoDB
dynamodb = [Link]("dynamodb")
table = [Link]("Orders")
table.put_item(Item={"order_id": "123", "status": "processed"})
6. Error Handling, Logging & Testing
Q32. How do you structure try/except blocks in a multi-step ETL job so a partial failure doesn't
corrupt downstream data?
Answer: Wrap each stage (extract, transform, load) in its own try/except so failures are isolated and
identifiable. Write to a staging location first and only move/commit to the final destination after the whole
batch succeeds (atomic 'swap' pattern), so a failure mid-load never leaves the destination in a half-written
state. Log enough context (batch id, record id) to resume or reprocess just the failed portion rather than the
whole job.
Q33. How do you implement idempotency in a Python ETL script so re-running it doesn't
duplicate data?
Answer: Common approaches: (1) use an upsert/merge (e.g., INSERT ... ON CONFLICT, or MERGE in
Redshift) keyed on a natural or surrogate key instead of plain INSERT/append; (2) write to a
uniquely-named output per run (e.g., partitioned by run date) and overwrite that partition rather than
appending; (3) track processed source file names/checksums in a manifest table and skip files already
processed.
Q34. What are the basics of pytest, and how would you unit test a transformation function?
# [Link]
def clean_amount(value):
if value is None:
return 0.0
return round(float(value), 2)
# test_transform.py
import pytest
from transform import clean_amount
def test_clean_amount_handles_none():
assert clean_amount(None) == 0.0
def test_clean_amount_rounds():
assert clean_amount(10.456) == 10.46
@[Link]("value,expected", [(5, 5.0), ("3.1", 3.1)])
def test_clean_amount_various(value, expected):
assert clean_amount(value) == expected
# Run with: pytest test_transform.py -v
7. Performance & Concurrency
Q35. When would you use multiprocessing vs threading vs asyncio in a data pipeline?
Answer: Multiprocessing: CPU-bound work (heavy parsing, transformations, compression) — bypasses the
GIL using separate processes. Threading: I/O-bound work with blocking libraries (e.g., requests, boto3 calls,
DB queries) — the GIL is released during I/O waits, so threads give real concurrency there. Asyncio:
I/O-bound work at high concurrency (thousands of lightweight network calls) using async-compatible
libraries (aiohttp, aioboto3) — more efficient than threads at scale since there's no per-thread OS overhead.
Q36. How would you parallelize downloading and processing 100 files from S3 in Python?
from [Link] import ThreadPoolExecutor
import boto3
s3 = [Link]("s3")
def process_file(key):
obj = s3.get_object(Bucket="my-bucket", Key=key)
data = obj["Body"].read()
# ... process data ...
return key
keys = [f"raw/file_{i}.csv" for i in range(100)]
with ThreadPoolExecutor(max_workers=16) as executor:
results = list([Link](process_file, keys))
Q37. What's the time/space complexity of the deduplication and grouping solutions shown
earlier?
Answer: dedupe_preserve_order: O(n) time (single pass, O(1) average set lookup) and O(n) space for the
seen set + result list. group_by: O(n) time and O(n) space for the grouped dict. dedupe_latest: O(n) time and
O(k) space where k is the number of distinct keys. merge_sorted ([Link]): O(n log k) time where k is
the number of input iterables, O(k) space since it's lazy/streaming rather than materializing the full merged
list.
Note: This guide was generated as a study aid. Verify any code snippets against current boto3/pandas/PySpark documentation
before using in production, and adapt examples to your actual schema and AWS environment.