EasyTechnical

22 practiced

Write a concise pandas expression to one-hot encode the categorical column df['color'] and drop the first level to avoid multicollinearity. Show how to keep the result in df (i.e., drop original column and add dummies). Also mention how to persist the encoding mapping for production inference to handle unseen categories.

Sample Answer

Use pandas.get_dummies with drop_first=True and assign back to df in one concise expression. For production, persist a fitted sklearn OneHotEncoder (handle_unknown='ignore') so you can transform new data with the same columns.

Pandas (quick, in-place):

python

# create dummies, drop original 'color', and concat to df
df = pd.concat(
    [df.drop('color', axis=1),
     pd.get_dummies(df['color'], prefix='color', drop_first=True)],
    axis=1
)

Why: pd.get_dummies(..., drop_first=True) produces k-1 columns to avoid multicollinearity (dummy trap). Concat after dropping original column keeps df consistent.

Production-safe approach (recommended):

python

from sklearn.preprocessing import OneHotEncoder
import joblib

# fit encoder on training data
enc = OneHotEncoder(drop='first', sparse=False, handle_unknown='ignore')
enc.fit(df[['color']])

# persist encoder
joblib.dump(enc, 'color_ohe.joblib')

# transform new data at inference time
ohe_cols = enc.get_feature_names_out(['color'])
X_ohe = pd.DataFrame(enc.transform(new_df[['color']]), columns=ohe_cols, index=new_df.index)
new_df = pd.concat([new_df.drop('color', axis=1), X_ohe], axis=1)

Notes:- handle_unknown='ignore' prevents errors on unseen categories (adds zero vector for unknowns).- Persisting the encoder (joblib) ensures consistent column order and encoding between train and inference.- Edge cases: nulls — decide to fillna('MISSING') before fitting if you want to treat NaNs as a category.

MediumTechnical

21 practiced

Implement a scikit-learn Pipeline that imputes numeric missing values with median, scales numeric features with StandardScaler, encodes categorical features using OneHotEncoder (handle unknowns), and trains a RandomForestClassifier. Show code using ColumnTransformer and Pipeline given lists numeric_cols and cat_cols. Also include how to validate this pipeline with GridSearchCV.

Sample Answer

Approach: build preprocessing with ColumnTransformer (median impute + scale for numerics; impute constant + OneHotEncoder for categoricals), compose into Pipeline with RandomForestClassifier, and tune with GridSearchCV.

python

from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import GridSearchCV

# assume numeric_cols and cat_cols are defined lists of column names
numeric_transformer = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler())
])

categorical_transformer = Pipeline([
    ("imputer", SimpleImputer(strategy="constant", fill_value="missing")),
    ("onehot", OneHotEncoder(handle_unknown="ignore", sparse=False))
])

preprocessor = ColumnTransformer([
    ("num", numeric_transformer, numeric_cols),
    ("cat", categorical_transformer, cat_cols)
], remainder="drop")

pipeline = Pipeline([
    ("preproc", preprocessor),
    ("clf", RandomForestClassifier(random_state=42, n_jobs=-1))
])

param_grid = {
    "clf__n_estimators": [100, 300],
    "clf__max_depth": [None, 10, 30],
    "clf__min_samples_leaf": [1, 4]
}

grid = GridSearchCV(pipeline, param_grid, cv=5, scoring="roc_auc", n_jobs=-1, verbose=1)
# Fit:
# grid.fit(X_train, y_train)
# Best model accessible via grid.best_estimator_
# Evaluate: grid.predict(X_test), grid.predict_proba(X_test)[:,1]

Key points:- OneHotEncoder(handle_unknown="ignore") prevents errors on unseen categories.- Imputing categoricals with a token keeps missing as a category.- Pipeline ensures preprocessing is applied inside CV folds (prevents leakage).Edge cases: very high-cardinality categorical cols (consider target/embedding encoding), extreme class imbalance (use stratified CV or class_weight).

MediumTechnical

25 practiced

Explain the bias-variance tradeoff. For a supervised ML model, describe what characteristic shapes you'd expect to see on training and validation learning curves for a high-bias model and a high-variance model. Then list practical changes you can implement in scikit-learn or PyTorch to address high variance and high bias respectively.

MediumTechnical

23 practiced

Explain how to save and load PyTorch model weights and optimizer state for checkpointing and resuming training. Include code snippets that show saving model.state_dict(), optimizer.state_dict(), and additional metadata (epoch, scheduler state) and explain device mapping when loading on CPU vs GPU.

Sample Answer

To checkpoint and resume PyTorch training reliably, save the model's parameters, optimizer state, (optionally) LR scheduler state and training metadata in a single dict using torch.save. On load, restore state_dicts and map tensors to the appropriate device.

Code — saving checkpoint:

python

import torch

def save_checkpoint(path, model, optimizer, scheduler=None, epoch=0, best_val=None):
    checkpoint = {
        "epoch": epoch,
        "model_state": model.state_dict(),
        "optimizer_state": optimizer.state_dict(),
        "best_val": best_val
    }
    if scheduler is not None:
        checkpoint["scheduler_state"] = scheduler.state_dict()
    torch.save(checkpoint, path)

Code — loading checkpoint (GPU available):

python

def load_checkpoint(path, model, optimizer=None, scheduler=None, device="cuda"):
    checkpoint = torch.load(path, map_location=device)
    model.load_state_dict(checkpoint["model_state"])
    if optimizer and "optimizer_state" in checkpoint:
        optimizer.load_state_dict(checkpoint["optimizer_state"])
        # If moving optimizer to GPU after load, ensure tensors are on correct device:
        for state in optimizer.state.values():
            for k, v in state.items():
                if isinstance(v, torch.Tensor):
                    state[k] = v.to(device)
    if scheduler and "scheduler_state" in checkpoint:
        scheduler.load_state_dict(checkpoint["scheduler_state"])
    epoch = checkpoint.get("epoch", None)
    return epoch, checkpoint.get("best_val", None)

Device mapping notes:- To load on CPU-only machine: use torch.load(path, map_location="cpu"). This maps all tensors to CPU.- To load a GPU-saved checkpoint onto a specific GPU: use map_location = lambda storage, loc: storage.cuda(desired_gpu_id).- Optimizer states contain tensors (e.g., momentum buffers). After loading on a different device, move these tensors to the target device as shown.- If model was wrapped with DataParallel/DistributedDataParallel, save model.module.state_dict() or handle prefixes when loading.

Best practices:- Save checkpoints atomically (save to tmp file then rename).- Include epoch, random seeds, and any preprocessing/config so experiments are reproducible.- For inference only, load model, call model.eval(), and you can skip optimizer/scheduler.

HardTechnical

18 practiced

Implement a memory-mapped data loader using numpy.memmap for a very large dataset stored on SSD that supports random sampling and minibatch reads. Describe a shuffling strategy that avoids loading the entire index into memory (e.g., block shuffling or indexed reservoir) and provide a code sketch showing how to open memmap, sample indices, and read minibatches efficiently while minimizing seeks.

Sample Answer

To handle very large datasets on SSD with numpy.memmap, use an out-of-core loader that:1) memory-maps the array, 2) samples indices without keeping full shuffled index in RAM (use block-shuffle or indexed-reservoir), and 3) reads minibatches with contiguous access to minimize SSD seeks (sort indices per read, prefetch nearby ranges).

Approach summary:- Partition the dataset into fixed-size blocks (e.g., 1M records or size tuned to SSD bandwidth).- For each epoch, randomly permute block order (cheap: few thousand blocks fits memory).- Within a block, either: - Fully shuffle indices if block fits RAM, or - Use an in-block reservoir/sample generator to stream random indices.- When producing minibatches: emit indices grouped by block, and for each minibatch sort by index to create near-sequential reads. Optionally prefetch next block.

Code sketch (Python + numpy):

python

import numpy as np
import math
import random

class MemmapLoader:
    def __init__(self, path, dtype, shape, block_size=1_000_000, batch_size=256):
        # shape = (N, feature_dim) or similar; we assume first dim is samples
        self.mm = np.memmap(path, dtype=dtype, mode='r', shape=shape)
        self.N = shape[0]
        self.block_size = block_size
        self.batch_size = batch_size
        self.blocks = math.ceil(self.N / block_size)

    def epoch_indices(self):
        # Shuffle block order
        block_order = list(range(self.blocks))
        random.shuffle(block_order)
        for b in block_order:
            start = b * self.block_size
            end = min(self.N, start + self.block_size)
            length = end - start
            # If block fits in memory, get a local shuffle
            if length <= 5_000_000:  # heuristic
                idx = np.arange(start, end)
                np.random.shuffle(idx)
                for i in range(0, length, self.batch_size):
                    yield idx[i:i+self.batch_size]
            else:
                # For very big blocks, do reservoir/random sampling stream
                # Here we create shuffled chunk indices to avoid full block shuffle
                chunk = 100_000
                chunks = math.ceil(length / chunk)
                chunk_order = list(range(chunks))
                random.shuffle(chunk_order)
                for c in chunk_order:
                    cs = start + c*chunk
                    ce = min(end, cs + chunk)
                    ids = np.arange(cs, ce)
                    np.random.shuffle(ids)
                    for i in range(0, len(ids), self.batch_size):
                        yield ids[i:i+self.batch_size]

    def __iter__(self):
        for batch_idx in self.epoch_indices():
            # minimize seeks: read contiguous runs by sorting (keeps locality)
            sorted_idx = np.sort(batch_idx)
            data = self.mm[sorted_idx]  # memmap slice triggers efficient read
            # reorder to original random order for the batch
            order = np.argsort(sorted_idx)
            # map sorted positions back to requested order
            inverse = np.argsort(order)  # simpler: use fancy indexing below
            # reconstruct in original order:
            # create mapping from sorted_idx position to original positions
            pos = {v:i for i,v in enumerate(sorted_idx)}
            orig_order = [pos[v] for v in batch_idx]
            batch = data[orig_order]
            yield batch

Key points:- Block-level shuffle keeps memory small: storing block ids is tiny.- Sorting indices per minibatch groups reads into contiguous ranges — reduces SSD seeks.- Tweak block_size/chunk to match SSD throughput and RAM.Complexity: I/O dominated; CPU cost for sorting O(B log B) per batch (B=batch_size). Edge cases: extremely small batch_size -> more seeks; very large feature dims -> mindful of memory for batch arrays. Optional improvements: async prefetching with a background thread/process, OS-level posix_fadvise or aio, use mmap with read-ahead, or store flattened contiguous records to reduce random access overhead.

Python Programming & ML Libraries Questions

Sample Answer

Sample Answer

Sample Answer

Sample Answer

Sample Answer

Unlock Full Question Bank