Basic ML Concepts in Code Questions

Introductory machine learning concepts explained with practical, code-oriented examples. Covers data preprocessing, model training loops, gradient descent optimization, loss functions, regularization, evaluation metrics, and common algorithms at a beginner-to-intermediate level to help engineers implement ML in real projects.

MediumTechnical

63 practiced

Create a minimal Flask API that loads a serialized scikit-learn Pipeline and serves batch predictions via a POST JSON endpoint. Include code for model loading at startup, request parsing, batching, and returning JSON responses, and describe thread-safety concerns.

Sample Answer

Approach: load a serialized scikit-learn Pipeline once at startup, accept a JSON array of feature dicts or arrays via POST /predict, run predictions in batches to control memory/latency, and return JSON with predictions. Use joblib for loading and Flask for the API. Discuss thread-safety and deployment notes.

python

# app.py
from flask import Flask, request, jsonify
import joblib
import numpy as np
from typing import List

MODEL_PATH = "model_pipeline.joblib"
BATCH_SIZE = 256  # tune based on memory / latency

app = Flask(__name__)

# Load model at import time (once per process)
model = joblib.load(MODEL_PATH)

def parse_payload(payload) -> List:
    """
    Accepts either:
    - {"instances": [[f1,f2,...], ...]}
    - {"instances": [{"feat1": v1, ...}, ...]}
    Returns list of feature rows suitable for model.predict.
    """
    if not isinstance(payload, dict) or "instances" not in payload:
        raise ValueError("Payload must be JSON object with key 'instances'")
    instances = payload["instances"]
    if len(instances) == 0:
        return []
    # If dicts, convert to list order using model.feature_names_in_ if available
    if isinstance(instances[0], dict):
        if hasattr(model, "feature_names_in_"):
            feature_order = list(model.feature_names_in_)
        else:
            feature_order = sorted(instances[0].keys())
        rows = [[inst.get(f, None) for f in feature_order] for inst in instances]
    else:
        rows = instances  # assume list of lists
    return rows

def batch_predict(rows, batch_size=BATCH_SIZE):
    preds = []
    for i in range(0, len(rows), batch_size):
        batch = rows[i : i + batch_size]
        X = np.array(batch)
        preds_batch = model.predict(X)
        preds.extend(preds_batch.tolist())
    return preds

@app.route("/predict", methods=["POST"])
def predict():
    try:
        payload = request.get_json(force=True)
        rows = parse_payload(payload)
    except Exception as e:
        return jsonify({"error": str(e)}), 400

    if len(rows) == 0:
        return jsonify({"predictions": []}), 200

    try:
        predictions = batch_predict(rows)
    except Exception as e:
        return jsonify({"error": "prediction error: " + str(e)}), 500

    return jsonify({"predictions": predictions}), 200

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

Thread-safety concerns and recommendations:- scikit-learn estimators are generally safe for concurrent reads (predict) but not for concurrent mutation. Loading model once per process (as above) is fine if you only call predict.- If using a multi-threaded server (Gunicorn with threads or Flask's threaded=True), ensure the model is not modified after load. If you must update the model at runtime, use process-level swapping (start new worker) or protect updates with a threading.Lock and ensure in-flight requests see a consistent object (atomic swap).- For heavy CPU-bound predict, prefer multiple worker processes (Gunicorn with multiple workers) instead of many threads to utilize multiple cores and avoid GIL contention.- For large models, consider memory copy cost: each worker process will have its own copy of the loaded model unless using copy-on-write-friendly loading before forking (load model before Gunicorn forks to share memory pages).- Add input validation, timeouts, and monitoring in production.

EasyTechnical

95 practiced

Write a minimal PyTorch training loop for one epoch that trains a provided nn.Module on a DataLoader. Include device handling (CPU/GPU), zeroing gradients, loss.backward, optimizer.step, model.train call, and basic logging of batch loss.

Sample Answer

Use a straightforward single-epoch training loop: move model and batches to device, set model.train(), zero grads, forward, compute loss, backward, optimizer.step(), and log batch loss.

python

import torch

def train_one_epoch(model: torch.nn.Module,
                    dataloader: torch.utils.data.DataLoader,
                    optimizer: torch.optim.Optimizer,
                    loss_fn,
                    device=None,
                    log_every=50):
    """
    Train model for one epoch over dataloader.
    - model: nn.Module
    - dataloader: yields (inputs, targets) or dicts/tensors
    - optimizer: optimizer with parameters
    - loss_fn: callable (preds, targets) -> loss tensor
    - device: 'cuda'/'cpu' or None to auto-select
    - log_every: frequency (batches) to print loss
    """
    device = device or ("cuda" if torch.cuda.is_available() else "cpu")
    model.to(device)
    model.train()

    running_loss = 0.0
    for batch_idx, batch in enumerate(dataloader, 1):
        # unpack batch (support tuple/list/dict)
        if isinstance(batch, (list, tuple)):
            inputs, targets = batch
        elif isinstance(batch, dict):
            inputs, targets = batch["inputs"], batch["targets"]
        else:
            raise ValueError("Dataloader must yield (inputs, targets) or dict")

        inputs = inputs.to(device)
        targets = targets.to(device)

        optimizer.zero_grad()             # zero gradients
        outputs = model(inputs)           # forward pass
        loss = loss_fn(outputs, targets)  # compute loss
        loss.backward()                   # backward pass
        optimizer.step()                  # update params

        running_loss += loss.item()
        if batch_idx % log_every == 0:
            avg = running_loss / log_every
            print(f"Batch {batch_idx}\tAvg Loss: {avg:.4f}")
            running_loss = 0.0

    # final small log if needed
    if running_loss > 0:
        print(f"Final batches avg loss: {running_loss / (batch_idx % log_every):.4f}")

Key points:- model.train() enables dropout/batchnorm training behavior.- Move both model and tensors to same device.- zero_grad before backward to avoid gradient accumulation.- Use loss.item() for logging to avoid retaining computation graph.- For production: consider mixed precision (torch.cuda.amp), gradient clipping, learning-rate schedulers, and handling variable batch structures.

Time complexity: O(N) per epoch (N = dataset size). Space: O(batch_size * sample_size) plus model params. Edge cases: empty dataloader, non-tensor batch elements, multiple targets—handle accordingly.

HardTechnical

47 practiced

Design and implement a lightweight model monitoring script in Python that runs alongside a serving process, computes rolling metrics such as latency distribution, request size, prediction distribution, and pushes those metrics to Prometheus via a /metrics endpoint or pushgateway. Include thresholds for alerting and discuss sampling strategies.

Sample Answer

Approach: run a lightweight Python monitoring sidecar (thread or process) that receives events from the serving process (e.g., via a local Unix socket, in-process queue, or HTTP POST), maintains rolling windows (time-based) for metrics, exposes Prometheus metrics via /metrics (prometheus_client) and supports pushing to Pushgateway. Use histograms for latency/request size and counters/gauges for distributions. Sample selectively under high throughput.

python

# python
import time, threading, collections, json
from prometheus_client import start_http_server, Counter, Gauge, Histogram, Summary, push_to_gateway

# Prometheus metrics
LATENCY_HIST = Histogram('model_latency_seconds', 'Inference latency seconds', buckets=(0.001,0.01,0.05,0.1,0.5,1,5))
REQ_SIZE_HIST = Histogram('request_size_bytes', 'Request payload size bytes', buckets=(100,500,1000,5000,10000))
PRED_COUNTER = Counter('prediction_label_total', 'Count of predictions', ['label'])
REQ_COUNTER = Counter('requests_total', 'Total requests')
ALERT_GAUGE = Gauge('monitor_alerts', '1 if alert triggered else 0', ['type'])

# Rolling window storage (time-bucketed)
WINDOW_SEC = 300
buckets = collections.deque()

def add_event(latency, req_size, prediction_label):
    now = time.time()
    if not buckets or now - buckets[-1]['t'] > 1:
        buckets.append({'t': now, 'events': []})
        # trim
        while buckets and now - buckets[0]['t'] > WINDOW_SEC:
            buckets.popleft()
    buckets[-1]['events'].append((latency, req_size, prediction_label))
    # update prometheus metrics
    LATENCY_HIST.observe(latency)
    REQ_SIZE_HIST.observe(req_size)
    REQ_COUNTER.inc()
    PRED_COUNTER.labels(label=str(prediction_label)).inc()

# Example consumer: simulate receiving events (replace with real ingestion)
def simulate_ingest():
    import random
    labels = ['A','B','C']
    while True:
        l = random.expovariate(10)
        s = random.randint(200,2000)
        lab = random.choice(labels)
        # sampling: only record 10% of low-latency events to reduce overhead
        if l>0.2 or random.random() < 0.1:
            add_event(l, s, lab)
        time.sleep(0.01)

# Alert evaluation
LATENCY_THRESHOLD = 1.0
ERROR_RATE_THRESHOLD = 0.05  # placeholder if error tracking added

def alert_loop():
    while True:
        # compute rolling 95th percentile approx from histogram buckets: use stored events for accuracy
        all_lat = [e[0] for b in buckets for e in b['events']]
        if all_lat:
            p95 = sorted(all_lat)[int(0.95*len(all_lat))-1]
            if p95 > LATENCY_THRESHOLD:
                ALERT_GAUGE.labels(type='latency_p95').set(1)
            else:
                ALERT_GAUGE.labels(type='latency_p95').set(0)
        time.sleep(10)

if __name__ == '__main__':
    start_http_server(8000)  # /metrics
    threading.Thread(target=simulate_ingest, daemon=True).start()
    threading.Thread(target=alert_loop, daemon=True).start()
    while True:
        time.sleep(60)
        # optional: push to gateway
        # push_to_gateway('pushgateway:9091', job='model_monitor', registry=None)

Key points:- Use time-bucketed deque for rolling windows to bound memory.- Use Histogram for latency/request size and Counters for prediction distribution.- Sampling: keep all high-latency/outlier events; sample low-latency events probabilistically to reduce cost.- Alerts: compute windowed stats (p95/p99) and set alert gauge; integrate with Prometheus Alertmanager via rules on exposed metrics.Edge cases: bursty traffic -> increase sampling; long-tail memory -> increase bucket granularity; clock skew -> use monotonic timestamps. Complexity: O(events) append; memory O(rate * WINDOW_SEC * sampling_rate).

HardTechnical

63 practiced

Write Python code to run a small hyperparameter sweep in parallel using concurrent.futures or multiprocessing. Sweep learning rate, weight decay, and batch size across a grid, run training jobs in parallel processes, collect results in a CSV, and pick the best setting. Discuss reproducibility and resource management concerns.

Sample Answer

Approach: use multiprocessing.Pool (or concurrent.futures.ProcessPoolExecutor) to run independent training jobs in parallel. Each job receives a hyperparameter tuple, sets deterministic seeds, runs a short training function (here simulated or using PyTorch), returns metrics. Collect results, save to CSV, pick best config. Manage GPU/CPU assignment, limit concurrency, and ensure reproducibility via seeds and deterministic settings.

python

import csv
import itertools
import os
import random
import time
from multiprocessing import Pool, cpu_count
import numpy as np

# Example train function (replace with real PyTorch/TensorFlow code)
def train_job(args):
    lr, wd, batch = args
    # Reproducibility: deterministic per-job seed
    seed = hash((lr, wd, batch)) & 0xffffffff
    random.seed(seed)
    np.random.seed(seed)
    # If using torch:
    # import torch
    # torch.manual_seed(seed); torch.use_deterministic_algorithms(True)
    # Simulate training (replace with model training + validation)
    time.sleep(0.5)  # simulate work
    # Simulated metric: higher lr may help up to a point, batch noisy
    val_acc = 0.7 + 0.2 * np.exp(-((np.log(lr)+3)**2)) - 0.01*wd + 0.02*(batch/256) + np.random.randn()*0.005
    return {"lr": lr, "weight_decay": wd, "batch_size": batch, "val_acc": float(val_acc)}

def main():
    lrs = [1e-4, 3e-4, 1e-3]
    wds = [0, 1e-4, 1e-3]
    batches = [32, 64, 128]
    grid = list(itertools.product(lrs, wds, batches))

    max_workers = min(len(grid), max(1, cpu_count() - 1))
    with Pool(processes=max_workers) as p:
        results = p.map(train_job, grid)

    # Write CSV
    out = "sweep_results.csv"
    with open(out, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=["lr","weight_decay","batch_size","val_acc"])
        writer.writeheader()
        writer.writerows(results)

    best = max(results, key=lambda r: r["val_acc"])
    print("Best config:", best)

if __name__ == "__main__":
    main()

Key points:- Reproducibility: set per-job seeds (random, numpy, torch), fix worker-level environment (PYTHONHASHSEED), and use deterministic ops where available. Log versions and git commit.- Resource management: limit concurrent processes to available CPUs/GPUs, avoid over-subscribing GPUs (assign CUDA_VISIBLE_DEVICES per process), monitor memory/CPU. For GPU sweeps, prefer one process per GPU or use job queue with device assignment.- Fault tolerance: checkpoint intermediate results, retry failed jobs, stream logs.- Scalability: for large sweeps, use job schedulers (Kubernetes, SLURM) or cloud hyperparameter services.

MediumTechnical

96 practiced

Write a TensorFlow tf.data pipeline in code that reads TFRecord files, parses serialized examples, shuffles, batches, applies a simple map transform, and prefetches for improved training throughput. Mention common performance knobs.

Sample Answer

Solution approach: use tf.data.TFRecordDataset to read files, a parse_fn with tf.io.parse_single_example, apply map transforms with num_parallel_calls=tf.data.AUTOTUNE, shuffle with an appropriate buffer_size, batch (or map_and_batch), and prefetch to overlap CPU and GPU work. Use AUTOTUNE and parallel interleave for scalability.

python

import tensorflow as tf

AUTOTUNE = tf.data.AUTOTUNE

# Example feature spec
feature_description = {
    "image_raw": tf.io.FixedLenFeature([], tf.string),
    "label": tf.io.FixedLenFeature([], tf.int64),
}

def _parse_example(serialized_example):
    # Parse the serialized TFRecord example
    example = tf.io.parse_single_example(serialized_example, feature_description)
    image = tf.io.decode_jpeg(example["image_raw"], channels=3)  # or decode_raw for raw bytes
    image = tf.image.convert_image_dtype(image, tf.float32)  # [0,1]
    label = tf.cast(example["label"], tf.int32)
    return image, label

def _simple_augment(image, label):
    # Simple map transform (data augmentation / preprocessing)
    image = tf.image.random_flip_left_right(image)
    image = tf.image.resize(image, [224, 224])
    return image, label

def make_dataset(tfrecord_files, batch_size=32, shuffle_buffer=10000, is_training=True):
    # Create dataset of filenames and read in parallel
    dataset = tf.data.TFRecordDataset(tfrecord_files, num_parallel_reads=AUTOTUNE)
    
    if is_training:
        dataset = dataset.shuffle(buffer_size=len(tfrecord_files))  # optional: shuffle filenames

    # Parse, transform, shuffle, batch, prefetch
    dataset = dataset.map(_parse_example, num_parallel_calls=AUTOTUNE)
    
    if is_training:
        dataset = dataset.shuffle(buffer_size=shuffle_buffer)  # shuffle examples
    
    dataset = dataset.map(_simple_augment, num_parallel_calls=AUTOTUNE)
    
    dataset = dataset.batch(batch_size, drop_remainder=True)
    dataset = dataset.prefetch(AUTOTUNE)  # overlap producer and consumer
    
    return dataset

# Usage
tfrecord_files = tf.io.gfile.glob("/data/train-*.tfrecord")
train_ds = make_dataset(tfrecord_files, batch_size=64, shuffle_buffer=20000, is_training=True)

Key points / performance knobs:- num_parallel_reads in TFRecordDataset and num_parallel_calls in map: set to tf.data.AUTOTUNE to maximize throughput.- shuffle buffer_size: larger gives better randomness but higher memory.- batch size and drop_remainder: affects GPU utilization and multi-replica sync.- prefetch(buffer_size=tf.data.AUTOTUNE): overlap CPU preprocessing and GPU training.- use cache() when dataset fits in memory to avoid re-reading.- use interleave/file sharding for many small files: tf.data.Dataset.list_files(...).interleave(...)- enable deterministic=False for faster nondeterministic pipeline where order doesn't matter.- consider map_and_batch, parallel_batch, and fused ops for further speedups.- For TPU/multi-GPU: shard dataset per replica, set drop_remainder=True.

Unlock Full Question Bank

Get access to hundreds of Basic ML Concepts in Code interview questions and detailed answers.

Join thousands of developers preparing for their dream job.