ML Engineer
Python & Software Engineering
Senior MLE roles require production-quality Python โ not just scripts. You need to understand Python's execution model, write clean OOP code, handle concurrency for data pipelines, and write testable, maintainable code. These are tested in coding rounds and system design discussions alike.
Core Theory
- 1Python GIL (Global Interpreter Lock): CPython allows only one thread to execute Python bytecode at a time. Threading is I/O-bound only โ while one thread waits for a network call, others run. For CPU-bound parallelism, use multiprocessing (separate processes, no GIL), or use NumPy/PyTorch operations which release the GIL. The GIL is invisible for ML workloads using NumPy/PyTorch because they run C extensions that release it.
- 2Generators and lazy evaluation: generators produce values one at a time on demand, never materializing the full sequence. Memory usage is O(1) regardless of data size. Critical for processing large ML datasets: instead of loading 10M rows into a list, yield one batch at a time. The yield keyword turns any function into a generator.
- 3Context managers (__enter__ / __exit__): resource management pattern. Examples: torch.no_grad() (disables gradient tracking), open() for files, database connections. Implement with class-based (__enter__/__exit__) or @contextmanager decorator. Ensures cleanup even if exceptions occur โ prevents GPU memory leaks in model serving.
- 4Decorators: functions that wrap other functions. Use cases in ML: @functools.lru_cache (memoize expensive feature computations), @retry (exponential backoff for API calls to external data sources), @timer (profiling), @dataclass (auto-generate __init__, __repr__, __eq__ for configuration objects).
- 5Type hints and dataclasses: from Python 3.10+, use type hints everywhere for IDE support, documentation, and runtime validation (pydantic). Dataclasses (@dataclass) auto-generate boilerplate. TypedDict for dict schemas. These signal production-quality code to interviewers.
- 6Concurrency models: threading (I/O-bound โ web scraping, API calls โ shares memory, limited by GIL), multiprocessing (CPU-bound โ feature computation, data preprocessing โ separate memory spaces, pickling overhead), asyncio (event loop โ coroutines for high-concurrency I/O without threads โ single-threaded, cooperative scheduling). For ML: preprocessing with multiprocessing, API serving with asyncio.
- 7Testing: pytest fixtures (reusable test components), parametrize (run same test with multiple inputs), monkeypatch (replace functions in tests). Mock external dependencies (database calls, API calls) with unittest.mock.patch. Property-based testing with Hypothesis generates random inputs to find edge cases. Always test data pipeline functions, not just models.
- 8Performance profiling: cProfile for function-level timing, line_profiler for line-level, memory_profiler for memory usage. Vectorize with NumPy before reaching for Cython or Numba. Rule: profile first, optimize the identified bottleneck. Premature optimization is the root of all evil in research code.
Key Formulas
| 1 | import time |
| 2 | import functools |
| 3 | import threading |
| 4 | from collections import OrderedDict |
| 5 | from contextlib import contextmanager |
| 6 | from dataclasses import dataclass |
| 7 | from typing import Iterator, TypeVar, Callable, Any |
| 8 | |
| 9 | T = TypeVar('T') |
| 10 | |
| 11 | # โโโ 1. Retry decorator with exponential backoff โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ |
| 12 | def retry(max_attempts: int = 3, backoff: float = 1.0, exceptions: tuple = (Exception,)): |
| 13 | """Decorator: retry a function with exponential backoff on failure.""" |
| 14 | def decorator(func: Callable) -> Callable: |
| 15 | @functools.wraps(func) # preserves func name, docstring |
| 16 | def wrapper(*args, **kwargs): |
| 17 | for attempt in range(max_attempts): |
| 18 | try: |
| 19 | return func(*args, **kwargs) |
| 20 | except exceptions as e: |
| 21 | if attempt == max_attempts - 1: |
| 22 | raise |
| 23 | wait = backoff * (2 ** attempt) |
| 24 | print(f" Attempt {attempt+1} failed: {e}. Retrying in {wait}s...") |
| 25 | time.sleep(wait) |
| 26 | return wrapper |
| 27 | return decorator |
| 28 | |
| 29 | @retry(max_attempts=3, backoff=0.5, exceptions=(ConnectionError,)) |
| 30 | def fetch_features(user_id: int) -> dict: |
| 31 | """Simulated API call that might fail.""" |
| 32 | if user_id % 3 == 0: # simulate flaky API |
| 33 | raise ConnectionError(f"Timeout for user {user_id}") |
| 34 | return {"user_id": user_id, "feature_x": user_id * 1.5} |
| 35 | |
| 36 | # โโโ 2. Thread-safe LRU cache (for feature caching in serving) โโโโโโโโโโโโโโโโ |
| 37 | class ThreadSafeLRUCache: |
| 38 | def __init__(self, capacity: int): |
| 39 | self.capacity = capacity |
| 40 | self.cache = OrderedDict() |
| 41 | self._lock = threading.Lock() |
| 42 | |
| 43 | def get(self, key: Any) -> Any: |
| 44 | with self._lock: |
| 45 | if key not in self.cache: |
| 46 | return None |
| 47 | self.cache.move_to_end(key) |
| 48 | return self.cache[key] |
| 49 | |
| 50 | def put(self, key: Any, value: Any) -> None: |
| 51 | with self._lock: |
| 52 | if key in self.cache: |
| 53 | self.cache.move_to_end(key) |
| 54 | self.cache[key] = value |
| 55 | if len(self.cache) > self.capacity: |
| 56 | self.cache.popitem(last=False) |
| 57 | |
| 58 | feature_cache = ThreadSafeLRUCache(capacity=10_000) |
| 59 | |
| 60 | # โโโ 3. Batch generator for memory-efficient dataset loading โโโโโโโโโโโโโโโโโโ |
| 61 | def batch_generator(data_path: str, batch_size: int = 32) -> Iterator[list]: |
| 62 | """Yield batches from a large file without loading it all into memory.""" |
| 63 | batch = [] |
| 64 | with open(data_path, 'r') if False else [f"row_{i}" for i in range(1000)] as rows: |
| 65 | for row in rows: |
| 66 | batch.append(row) |
| 67 | if len(batch) == batch_size: |
| 68 | yield batch |
| 69 | batch = [] |
| 70 | if batch: # yield final partial batch |
| 71 | yield batch |
| 72 | |
| 73 | # โโโ 4. Context manager for timing (profiling in research) โโโโโโโโโโโโโโโโโโโ |
| 74 | @contextmanager |
| 75 | def timer(label: str = ""): |
| 76 | start = time.perf_counter() |
| 77 | try: |
| 78 | yield |
| 79 | finally: |
| 80 | elapsed = time.perf_counter() - start |
| 81 | print(f" [{label}] {elapsed*1000:.2f}ms") |
| 82 | |
| 83 | # โโโ 5. Dataclass for model config (clean alternative to dict) โโโโโโโโโโโโโโโโ |
| 84 | @dataclass |
| 85 | class ModelConfig: |
| 86 | learning_rate: float = 1e-3 |
| 87 | batch_size: int = 32 |
| 88 | n_epochs: int = 10 |
| 89 | dropout: float = 0.1 |
| 90 | model_name: str = "baseline-v1" |
| 91 | |
| 92 | def __post_init__(self): |
| 93 | assert 0 < self.learning_rate < 1, "LR must be in (0, 1)" |
| 94 | assert self.batch_size > 0 |
| 95 | |
| 96 | # โโโ Demo โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ |
| 97 | cfg = ModelConfig(learning_rate=0.01, batch_size=64, model_name="experiment-v2") |
| 98 | print(f"Config: {cfg}") # auto-generated __repr__ |
| 99 | |
| 100 | feature_cache.put("user_42", {"age": 28, "lifetime_value": 450.0}) |
| 101 | result = feature_cache.get("user_42") |
| 102 | print(f"Cached features: {result}") |
| 103 | |
| 104 | with timer("feature_lookup"): |
| 105 | _ = [feature_cache.get(f"user_{i}") for i in range(1000)] |
Five production Python patterns used in ML engineering: (1) retry decorator with exponential backoff for flaky external API calls; (2) thread-safe LRU cache with a lock for concurrent feature serving; (3) batch generator that never loads the full dataset into memory; (4) context manager for profiling code blocks; (5) dataclass for model configs with automatic validation.
Worked Interview Problems
3 problemsProblem
Write a PyTorch Dataset class that: (1) loads data lazily (not all into RAM), (2) supports caching processed samples, (3) handles errors gracefully. Design for a 10TB dataset.
Solution
Lazy loading: store file paths in __init__, not data. In __getitem__, load the file, preprocess, and return. This means only one sample is in memory at a time (O(1) memory).
Caching: add an optional disk cache (functools.lru_cache for in-memory or joblib.Memory for disk). First access: load, preprocess, cache. Subsequent access: load from cache. Cache key: file path + preprocessing config hash.
Error handling: wrap __getitem__ in try/except. On error, log the sample path and return the next valid sample, or return None and handle in the DataLoader's collate_fn. Avoid crashing the entire training run on one corrupt sample.
Multiprocessing safety: if DataLoader uses num_workers > 0, each worker is a separate process. Don't store file handles or database connections in __init__ โ they're not picklable. Open and close in __getitem__.
Progress monitoring: add a __len__ method. Log every 1000 samples with a callback. For distributed training, DistributedSampler handles which samples each GPU processes.
Answer
Key design: store file paths in __init__ (not data), load lazily in __getitem__, wrap in try/except for corrupted samples, avoid non-picklable resources for multiprocess DataLoader. Add disk cache with joblib for repeated preprocessing passes.
Common Mistakes
Using mutable default arguments in function definitions: def func(data=[]) โ the list is created once and shared across all calls. Use def func(data=None): if data is None: data = [] instead.
Not using functools.wraps when writing decorators โ without it, the wrapped function loses its __name__, __doc__, and other attributes, breaking introspection and logging.
Opening file handles or database connections in __init__ of a PyTorch Dataset โ they're not picklable and will fail when DataLoader spawns worker processes. Always open/close connections in __getitem__.
Using time.sleep() for testing async code โ it makes tests slow and flaky. Use unittest.mock.AsyncMock or pytest-asyncio fixtures to properly test async functions.
Forgetting to release GPU memory in serving code โ in PyTorch, use torch.cuda.empty_cache() after processing each batch to prevent OOM when handling variable-length inputs in a long-running server.