Python Programming & ML Libraries Questions

Python programming language fundamentals (syntax, data structures, control flow, error handling) with practical usage of machine learning libraries such as NumPy, pandas, scikit-learn, TensorFlow, and PyTorch for data manipulation, model development, training, evaluation, and lightweight ML tasks.

EasyTechnical

25 practiced

Using pandas, write a Python snippet that computes the rolling 7-day average of column 'value' per user in a DataFrame with columns ['user_id', 'timestamp', 'value']. Show how to: (1) ensure rolling uses a time window (7 days) not a fixed row count, (2) handle missing days per user, and (3) return the result aligned to the original timestamps. Include a brief example DataFrame and the code you would run.

Sample Answer

Approach: convert timestamp to datetime, sort by user and time, then use a time-based window with rolling('7D') applied per-user. Two options:- Direct per-group rolling on a datetime index (uses a 7-day lookback from each timestamp and handles irregular sampling).- If you want explicit missing days (e.g., ensure daily continuity before rolling), resample each user to a daily index, compute the rolling average, then re-align back to the original timestamps.

Example DataFrame and code:

python

import pandas as pd
import numpy as np

# Example data
df = pd.DataFrame({
    "user_id": [1,1,1,2,2],
    "timestamp": [
        "2025-11-20 10:00", "2025-11-22 09:00", "2025-11-28 12:00",
        "2025-11-21 08:00", "2025-11-27 18:00"
    ],
    "value": [10, 20, 30, 5, 15]
})
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.sort_values(["user_id","timestamp"])

Option A — time-window rolling per user (aligned to original timestamps):

python

def rolling_7d_per_user(g):
    # set timestamp as index so rolling('7D') uses time window
    g = g.set_index("timestamp").sort_index()
    g["rolling_7d"] = g["value"].rolling("7D").mean()
    return g.reset_index()

result_a = df.groupby("user_id", group_keys=False).apply(rolling_7d_per_user)
# result_a has rolling_7d aligned to original timestamps

Option B — ensure missing days are considered (resample to daily, then reindex back):

python

def resample_roll_realign(g):
    g = g.set_index("timestamp").sort_index()
    # resample to daily frequency to make missing days explicit
    daily = g["value"].resample("D").mean()   # NaN on days without observations
    daily_roll = daily.rolling("7D").mean()
    # bring rolling values back to original timestamps by reindexing
    aligned = daily_roll.reindex(g.index, method=None)  # no forward/backfill
    g["rolling_7d"] = aligned.values
    return g.reset_index()

result_b = df.groupby("user_id", group_keys=False).apply(resample_roll_realign)

Key points:- rolling("7D") uses a time-based 7-day window (not fixed row count).- Option A works for irregular timestamps and computes mean over the previous 7 days of actual samples.- Option B forces daily continuity so days without measurements are explicit (NaN) and influence the window as gaps; useful if you need daily-aware averaging.- Both results remain aligned to the original timestamps; choose A or B depending on whether missing days should be treated as absent (A) or explicit gaps (B).

Complexity: O(n) per group for rolling; memory depends on grouping and any resampling. Edge cases: single observation users, duplicate timestamps, timezone-aware datetimes.

MediumTechnical

20 practiced

Write an async Python function using aiohttp and asyncio that concurrently downloads a list of URLs, limits concurrency to N simultaneous requests using a semaphore, reuses a single ClientSession, handles timeouts and cancellations gracefully, and writes responses to disk. Explain when async I/O is preferable for ML pipelines such as fetching many small files from remote stores.

Sample Answer

Approach: use a single aiohttp.ClientSession, an asyncio.Semaphore to limit concurrency to N, aiohttp timeouts, and aiofiles to write responses without blocking the event loop. Handle cancellations and per-request failures robustly and ensure session closed.

python

import asyncio
import aiohttp
import aiofiles
from pathlib import Path

async def fetch_and_save(url: str, session: aiohttp.ClientSession, sem: asyncio.Semaphore, out_dir: Path, timeout: float = 10.0):
    filename = out_dir / (url.replace("://", "_").replace("/", "_")[:200] + ".bin")
    try:
        async with sem:
            # per-request timeout
            async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout)) as resp:
                resp.raise_for_status()
                # stream to avoid large-memory buffering
                async with aiofiles.open(filename, "wb") as f:
                    async for chunk in resp.content.iter_chunked(1024 * 32):
                        await f.write(chunk)
        return (url, "ok")
    except asyncio.CancelledError:
        # propagate cancellation for outer loop to handle
        raise
    except Exception as e:
        # log or collect errors
        return (url, f"error: {type(e).__name__}: {e}")

async def download_urls(urls, out_dir="downloads", concurrency=10, timeout=10.0):
    out_dir = Path(out_dir)
    out_dir.mkdir(parents=True, exist_ok=True)
    sem = asyncio.Semaphore(concurrency)

    # reuse single session
    conn = aiohttp.TCPConnector(limit=0)  # session-level connection pool, 0 means default unlimited per host
    async with aiohttp.ClientSession(connector=conn) as session:
        tasks = [asyncio.create_task(fetch_and_save(u, session, sem, out_dir, timeout)) for u in urls]
        try:
            results = await asyncio.gather(*tasks, return_exceptions=False)
        except asyncio.CancelledError:
            # if caller cancels, propagate cancel to all tasks and await to finish cleanup
            for t in tasks:
                t.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)
            raise
    return results

Key points:- Semaphore bounds concurrent requests to "concurrency".- Single ClientSession improves connection reuse and performance.- Streaming response and aiofiles avoid storing whole response in memory.- Timeout via aiohttp.ClientTimeout enforces per-request limits.- Cancellation handling cancels tasks and cleans up.

Complexity: IO-bound; wall-clock time ~ O(total_io / concurrency). Space: O(max in-flight buffers) ~ O(concurrency * chunk_size).

When async I/O is preferable for ML pipelines:- Ideal when fetching many small files/metadata (high-latency network waits) because async lets you overlap network waits without dedicating threads. This reduces resource usage (fewer threads/processes) and increases throughput for many concurrent small downloads (common in dataset sharding, feature fetch, prefetching). For large single-file GPU-heavy workloads, multi-threaded or multiprocessing with optimized transfer (parallel multipart, cloud SDKs) may be better. Use async when tasks are IO-bound and you need high concurrency with low memory/CPU overhead.

HardTechnical

18 practiced

List and explain all steps to make a PyTorch training job as reproducible as possible across runs and machines. Include seeding for Python, NumPy, and torch, setting torch.backends.cudnn.deterministic and torch.backends.cudnn.benchmark flags, worker_init_fn for DataLoader, mapping loaded models to device, and remaining sources of nondeterminism. Provide code snippets and discuss performance trade-offs.

Sample Answer

1) Set global seeds (Python, NumPy, torch, CUDA)- Rationale: ensure RNGs start from same state across runs/machines.- Code:

python

import os, random
import numpy as np
import torch

def set_seed(seed: int):
    os.environ['PYTHONHASHSEED'] = str(seed)
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)  # if using multiple GPUs

2) Configure cuDNN for determinism- Rationale: cuDNN uses nondeterministic algorithms by default for speed.- Code:

python

torch.backends.cudnn.deterministic = True
torch.backends.cudnn.benchmark = False

- Trade-off: deterministic=True may force slower algorithms and disable pruning of algorithm search (benchmark=False), reducing performance.

3) DataLoader reproducibility (worker seeds)- Rationale: each worker must have deterministic RNGs.- Code:

python

def worker_init_fn(worker_id):
    worker_seed = torch.initial_seed() % 2**32
    np.random.seed(worker_seed)
    random.seed(worker_seed)

loader = torch.utils.data.DataLoader(dataset, num_workers=4,
                                     worker_init_fn=worker_init_fn,
                                     shuffle=True)

4) Map loaded models/tensors to device deterministically- Rationale: ensure model parameters and optimizer states use same device; when loading checkpoints, call map_location.- Code:

python

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
state = torch.load(path, map_location=device)
model.load_state_dict(state['model'])
model.to(device)

5) Deterministic operations and known nondeterministic ops- Replace known nondeterministic ops (e.g., torch.nn.functional.dropout is deterministic with seed, but some fused ops, atomic reductions, certain CUDA kernels, and torch.nonzero order) or use deterministic fallback implementations.- Check torch.use_deterministic_algorithms(True) to enforce:

python

torch.use_deterministic_algorithms(True)

This will raise on nondeterministic ops. Trade-off: may break or be slower.

6) Other sources of nondeterminism- Parallelism: different number of threads (set OMP_NUM_THREADS, MKL_NUM_THREADS).- Different hardware or driver/CUDA/cuDNN versions — ensure same software stack.- Floating-point non-associativity across devices or distributed training (reduce ops).- Asynchronous CUDA operations; call torch.cuda.synchronize() when measuring.- Data shuffling across distributed nodes — use torch.distributed with seeded samplers:

python

sampler = torch.utils.data.DistributedSampler(dataset, seed=seed)

7) Additional best practices- Log seed and full environment (python, torch, CUDA/cuDNN versions, GPU model).- Disable nondeterministic optimizations or mixed precision if it introduces nondeterminism (AMP can be deterministic but test).- Save and restore optimizer RNG states: torch.get_rng_state(), torch.cuda.get_rng_state_all().

Performance trade-offs summary:- Deterministic settings and use_deterministic_algorithms(True) often reduce throughput and increase memory; cuDNN deterministic algorithms can be slower. DataLoader with fewer worker nondeterminism risks slower data pipelines. You must balance reproducibility vs training speed; for final experiments where exact bitwise reproducibility matters, accept the slowdown.

MediumTechnical

19 practiced

Implement a custom PyTorch Dataset class in Python that reads image file paths and labels from a CSV file. The Dataset must implement __len__ and __getitem__, lazily open images, apply torchvision transforms passed in as an argument, and handle corrupted images gracefully by logging and returning a fallback tensor. Provide code and discuss performance considerations when used with DataLoader num_workers > 0.

Sample Answer

Here's a concise implementation and discussion.

To implement this Dataset:1. Read CSV paths/labels in __init__ (don't open images).2. Lazily open images in __getitem__; apply transforms.3. Catch image-open/process exceptions, log them, and return a fallback tensor and label.

python

import csv
import logging
from pathlib import Path
from typing import Callable, Optional, List, Tuple

import torch
from torch.utils.data import Dataset
from PIL import Image, ImageFile

ImageFile.LOAD_TRUNCATED_IMAGES = True  # be more tolerant of truncated files

logger = logging.getLogger(__name__)

class CSVDataset(Dataset):
    def __init__(self, csv_path: str, root_dir: Optional[str] = None,
                 transform: Optional[Callable] = None,
                 fallback_size: Tuple[int,int]=(3,224,224)):
        """
        csv file: each row [relative_or_absolute_path,label]
        root_dir: optional base directory for relative paths
        transform: torchvision transforms to apply (should accept PIL Image)
        fallback_size: tensor shape (C,H,W) for fallback image
        """
        self.root = Path(root_dir) if root_dir else None
        self.transform = transform
        self.fallback_size = fallback_size
        self.samples: List[Tuple[str,int]] = []

        with open(csv_path, newline='') as f:
            reader = csv.reader(f)
            for row in reader:
                if not row:
                    continue
                path, label = row[0], int(row[1])
                self.samples.append((path, label))

    def __len__(self):
        return len(self.samples)

    def _make_fallback(self):
        # deterministic fallback (zeros) — could be mean image instead
        return torch.zeros(self.fallback_size, dtype=torch.float32)

    def __getitem__(self, idx: int):
        path, label = self.samples[idx]
        img_path = Path(path) if Path(path).is_absolute() or self.root is None else self.root / path
        try:
            # Lazily open image here
            with Image.open(img_path) as img:
                img = img.convert('RGB')
                if self.transform:
                    img = self.transform(img)  # transforms typically produce tensor
                else:
                    # fallback conversion if no transform provided
                    img = torch.from_numpy(np.array(img)).permute(2,0,1).float() / 255.0
            return img, label
        except Exception as e:
            logger.exception("Failed to load/process image %s: %s", img_path, e)
            return self._make_fallback(), label

Key points and reasoning:- Lazily opening images in __getitem__ avoids keeping many file handles open and reduces memory.- Using a try/except prevents a single corrupted image from crashing training; logging with logger.exception captures stack traces.- Returning a deterministic fallback tensor ensures batch collate shapes remain consistent.

Performance considerations with DataLoader(num_workers > 0):- Dataset must be picklable because workers are separate processes; avoid unpicklable state in __init__ (open file handles, threads).- Each worker has its own copy of Dataset; reading CSV into a small in-memory list is fine. For very large lists, consider memory mapping or shared indices.- I/O-bound workload benefits from multiple workers — use num_workers to overlap file reads and transforms. But too many workers cause file system contention (small files), CPU contention during heavy transforms, or increased RAM pressure.- If transforms are CPU-heavy (augmentation, complex resizing), prefer using more workers or move augmentations to GPU (nb: that requires different pipeline).- Use pin_memory=True when transferring tensors to GPU to speed up host->device copy.- Ensure any random state in transforms is worker-safe; set torch.utils.data.get_worker_info() in __getitem__ or use worker_init_fn to seed RNGs per worker.- For robustness: consider warming caches, batching small files into LMDB, or using webdataset if filesystem becomes a bottleneck.

Edge cases: ensure CSV indexing is valid, label parsing robust, and fallback tensor shape matches model input.

MediumTechnical

20 practiced

Describe NumPy broadcasting rules in detail and illustrate with examples aligning arrays of shapes (3,1) and (1,4) to produce (3,4). Explain how broadcasting works with trailing axes, when dimensions must be 1 or equal, and provide tips to avoid accidental creation of huge temporaries that can blow memory. Include guidance on using np.expand_dims and np.broadcast_to to make intent explicit.

Sample Answer

NumPy broadcasting rules (concise):- Align shapes from trailing (rightmost) axes.- For each axis, sizes are compatible if they are equal or one of them is 1.- If one array has fewer dimensions, it is virtually left-padded with 1s.- The resulting shape is the elementwise maximum of input sizes per axis.

Concrete example: shapes (3,1) and (1,4) -> result (3,4)- Align trailing axes: (3,1) (1,4)- Axis-by-axis: - Last axis: 1 and 4 -> compatible -> result 4 - Second-last: 3 and 1 -> compatible -> result 3- NumPy effectively replicates the size-1 axes to match the other array: - (3,1) behaves like (3,4) by repeating columns - (1,4) behaves like (3,4) by repeating rows

Example code:

python

import numpy as np
a = np.arange(3).reshape(3,1)    # shape (3,1)
b = np.arange(4).reshape(1,4)    # shape (1,4)
c = a + b                        # shape (3,4) via broadcasting

Notes on trailing axes and the "1 or equal" rule:- Always compare from the right. Example: (2,3,1) and (3,1,4) -> pad and compare right-to-left.- A dimension must be either 1 or equal to the other array’s size; otherwise broadcasting raises ValueError.

Memory and performance tips (avoid huge temporaries):- Broadcasting is usually a view/virtual replication and does not allocate the full repeated array. However, some operations produce temporaries (e.g., creating an explicit tiled array).- Avoid np.tile unless you need a real copy; prefer broadcasting expressions (a + b) which are lazy in shape but may still allocate the result.- Use in-place operations when possible (e.g., out= parameter) to avoid extra allocation.- For read-only explicit view without copying, use np.broadcast_to:

python

A_b = np.broadcast_to(a, (3,4))   # view-like; may be read-only

- Use np.expand_dims or np.newaxis to make intent explicit rather than relying on implicit shape-padding:

python

a = np.arange(3)[:, np.newaxis]   # (3,1)
b = np.arange(4)[np.newaxis, :]   # (1,4)

- If an operation would create a huge temporary (e.g., broadcasting to very large shape), consider: - chunking the computation, - using memory-efficient alternatives (np.einsum, matrix ops) that avoid explicit expansion, - performing reductions before expanding dimensions.

Summary guidance:- Remember right-align, 1-or-equal rule, and maximum size per axis.- Use expand_dims/np.newaxis for clarity; use np.broadcast_to when you need an explicit (non-copy) view; avoid np.tile and unnecessary intermediate arrays; and prefer in-place or chunked computations to prevent blowing memory.

Unlock Full Question Bank

Get access to hundreds of Python Programming & ML Libraries interview questions and detailed answers.

Join thousands of developers preparing for their dream job.