Python Coding and Data Structures Questions

Proficiency in Python, including arrays, dictionaries, linked lists, and basic algorithms. Ability to write efficient, clean code under time pressure. Understanding of time/space complexity and optimization.

HardTechnical

31 practiced

Implement a simple mark-and-sweep garbage collector simulation in Python. Represent objects and their references with a dict of id -> list_of_child_ids, implement the mark phase from given roots, then sweep unreachable objects. Explain how this differs from reference counting and provide complexity analysis.

Sample Answer

Approach: Use DFS/BFS from given roots to mark reachable object ids, then sweep the heap dict removing unmarked ids. This simulates mark-and-sweep: two phases, safe for cycles (unlike naive reference counting).

python

def mark_and_sweep(heap, roots):
    """
    heap: dict[id] -> list_of_child_ids
    roots: iterable of root ids
    Returns: (marked_set, new_heap) where new_heap contains only reachable objects
    """
    marked = set()
    stack = list(roots)

    # Mark phase: iterative DFS to avoid recursion limits
    while stack:
        obj = stack.pop()
        if obj in marked:
            continue
        if obj not in heap:
            # root refers to external/non-existing object; skip
            continue
        marked.add(obj)
        # push children; ignore missing child ids gracefully
        for child in heap.get(obj, []):
            if child not in marked:
                stack.append(child)

    # Sweep phase: remove unmarked objects
    new_heap = {obj: children for obj, children in heap.items() if obj in marked}
    return marked, new_heap

# Example
if __name__ == "__main__":
    heap = {
      1: [2, 3],
      2: [4],
      3: [2],
      4: [],
      5: [6],   # unreachable cycle 5 <-> 6
      6: [5]
    }
    roots = [1]
    marked, new_heap = mark_and_sweep(heap, roots)
    print("marked:", marked)      # {1,2,3,4}
    print("new_heap:", new_heap)  # only keys 1,2,3,4

Key points:- Correctly handles cycles because reachability is computed from roots.- Handles missing child ids and extraneous roots safely.- Can be implemented with BFS/DFS; iterative DFS used to avoid recursion depth issues.

Complexity:- Time: O(N + E) where N = number of objects, E = total references (each object and edge visited once).- Space: O(N) for mark set and stack (worst case).

How it differs from reference counting:- Reference counting reclaims objects when refcount drops to zero, O(1) per update, but fails to collect cyclic garbage (objects referencing each other). Mark-and-sweep periodically pauses to compute global reachability and reclaims cycles.- Mark-and-sweep has higher pause/work proportional to heap size; reference counting distributes cost but needs cycle detection/compaction for the same guarantees.

HardTechnical

31 practiced

Design an efficient Python approach to detect schema anomalies across millions of JSON documents representing event records. Your solution should stream documents, aggregate observed types per field path, and report fields with type changes, missingness, or structural drift. Provide prototype code and discuss accuracy versus sampling trade-offs.

Sample Answer

Approach: stream JSON documents, walk each record to emit canonical field paths (dot notation, arrays with []), and maintain an incremental sketch per path: counts, observed types (set or small-count map), presence count, and sample examples. Use reservoir sampling per path to store examples. Periodically compute anomalies: new types introduced, type flips over time, high missingness, nested structural drift (object → primitive, array→object), and rate-of-change alerts.

python

import json, sys, math, random
from collections import defaultdict, Counter

def iter_paths(obj, prefix=""):
    if isinstance(obj, dict):
        for k,v in obj.items():
            yield from iter_paths(v, f"{prefix}.{k}" if prefix else k)
    elif isinstance(obj, list):
        # treat arrays uniformly; capture element path under "field[]"
        for i,el in enumerate(obj):
            yield from iter_paths(el, f"{prefix}[]")
        if not obj:
            yield (prefix+"[]", "empty_array")
    else:
        yield (prefix, type(obj).__name__)

class PathStat:
    def __init__(self, sample_size=50):
        self.count = 0
        self.types = Counter()
        self.sample = []
        self.sample_size = sample_size
    def observe(self, typ, raw):
        self.count+=1
        self.types[typ]+=1
        if len(self.sample) < self.sample_size:
            self.sample.append(raw)
        else:
            # reservoir sampling
            r = random.randint(0, self.count-1)
            if r < self.sample_size:
                self.sample[r]=raw

def detect(stream):
    stats = defaultdict(PathStat)
    total=0
    for line in stream:
        total+=1
        doc = json.loads(line)
        for path,typ in iter_paths(doc):
            stats[path].observe(typ, None)
        # optionally emit periodic summaries
        if total%100000==0:
            yield ("checkpoint", total)
    yield ("final", stats, total)

# consumer would compute anomalies from stats: types with low frequency, flips over time, missingness = 1 - (count/total)

Key points:- Streaming avoids loading all data; per-path state is O(number of active paths) which typically << documents.- Use reservoir sampling to keep representative examples while bounded memory.- Track time-windowed stats (e.g., per-hour) to detect temporal drifts and sudden type flips.- Complexity: O(N * avg_fields_per_doc) time, memory O(P * s) where P is distinct paths, s sample size.

Accuracy vs sampling trade-offs:- Full counting (no sampling) yields exact rates but may be memory-heavy for many rare paths. Reservoir sampling keeps bounded memory and still captures rare-but-important examples probabilistically; detection sensitivity for very rare anomalies decreases.- Use adaptive sampling: allocate larger samples to high-change or high-value paths; use bloom filters + counters to cheaply detect new paths.- For critical pipelines, combine lightweight streaming detection for coverage and targeted full scans on flagged partitions for high-accuracy confirmation.

MediumTechnical

23 practiced

Implement serialize(root) and deserialize(data) functions in Python for binary trees with integer node values. Use preorder traversal with explicit null markers or level-order encoding. Ensure the format handles negative numbers and duplicates and is compact and reversible.

Sample Answer

Approach: use preorder traversal with an explicit null marker ("#") and comma separators. Preorder is compact, handles negatives and duplicates (values serialized as their string forms), and is easy to parse deterministically for deserialization.

python

# Definition for a binary tree node.
class TreeNode:
    def __init__(self, val=0, left=None, right=None):
        self.val = val
        self.left = left
        self.right = right

def serialize(root):
    """Serialize tree to a compact string using preorder with null markers."""
    parts = []
    def dfs(node):
        if node is None:
            parts.append('#')
            return
        parts.append(str(node.val))
        dfs(node.left)
        dfs(node.right)
    dfs(root)
    return ','.join(parts)

def deserialize(data):
    """Deserialize string back to tree produced by serialize."""
    if not data:
        return None
    tokens = iter(data.split(','))
    def build():
        try:
            t = next(tokens)
        except StopIteration:
            return None
        if t == '#':
            return None
        node = TreeNode(int(t))
        node.left = build()
        node.right = build()
        return node
    return build()

Key points:- Uses '#' to mark nulls so structure is preserved even with duplicates/negatives.- parse uses int() so negative numbers handled.Time complexity: O(n) for both serialize and deserialize (n = nodes). Space: O(n) for output and recursion stack O(h).Edge cases: empty tree (returns empty string or handled if empty input), malformed data (assumes valid input). Alternative: level-order (BFS) encoding if explicit breadth representation preferred; preorder is simpler and compact.

HardTechnical

32 practiced

Design and implement a RandomizedSet class in Python supporting insert(val), remove(val), and get_random() in average O(1) time. Ensure get_random returns a uniformly random element among current elements. Discuss how you would extend this to support duplicates and weighted sampling.

Sample Answer

To get O(1) average for insert, remove, and uniform get_random, store elements in a list for O(1) random access and a hashmap mapping value → index in the list for O(1) lookup. On remove, swap target with last element, pop, and update the moved element's index.

python

import random

class RandomizedSet:
    def __init__(self):
        self.arr = []            # list of values
        self.pos = {}            # val -> index in arr

    def insert(self, val):
        if val in self.pos:
            return False
        self.pos[val] = len(self.arr)
        self.arr.append(val)
        return True

    def remove(self, val):
        if val not in self.pos:
            return False
        idx = self.pos[val]
        last = self.arr[-1]
        # move last into idx (unless removing last)
        self.arr[idx] = last
        self.pos[last] = idx
        # remove last
        self.arr.pop()
        del self.pos[val]
        return True

    def get_random(self):
        if not self.arr:
            raise IndexError("empty set")
        return random.choice(self.arr)

Key points:- insert/remove average O(1), get_random O(1).- Uniformity guaranteed by random.choice over list.

Extensions:- Allow duplicates: change pos to map val → set/list of indices, and when removing pick an index from that set; still swap last element and update its index entries. This becomes slightly more bookkeeping but remains amortized O(1).- Weighted sampling: maintain an array of elements with repetition proportional to weight (fast but memory heavy), or better, use Alias Method (O(1) sampling after O(n) preprocess) or binary-search on prefix-sum array (O(log n) sampling, O(1) update not trivial). For dynamic weights with frequent updates, use a Fenwick tree (BIT) over weights to support O(log n) update and O(log n) sample by prefix-sum.

MediumTechnical

29 practiced

You must compute sum(value) grouped by user_id from a CSV file too large to fit in memory. Using only Python standard libraries, write a streaming approach that produces aggregated sums per user with bounded memory footprint, or explain when external aggregation (partition + reduce) is required. Provide code sketch and complexity analysis.

Sample Answer

Approach: If the number of distinct user_id fits in memory, stream the CSV line-by-line and maintain a dict user_id → running sum (bounded by number of users). If distinct user_ids are too many, use external aggregation: hash-partition the file into N smaller temporary files (based on user_id % N), then stream each partition and aggregate (reduce) separately. That guarantees bounded memory per partition.

Code sketch (Python standard library):

python

import csv
import tempfile
import os
from collections import defaultdict

def stream_aggregate(csv_path):
    # Single-pass in-memory aggregator (works if unique users fit in memory)
    sums = defaultdict(float)
    with open(csv_path, newline='') as f:
        reader = csv.DictReader(f)
        for row in reader:
            uid = row['user_id']
            val = float(row['value'])
            sums[uid] += val
    return sums  # dict of aggregated sums

def external_aggregate(csv_path, num_partitions=64):
    # Partition phase
    temps = [tempfile.NamedTemporaryFile(mode='w+', delete=False, newline='') for _ in range(num_partitions)]
    writers = [csv.writer(t) for t in temps]
    with open(csv_path, newline='') as f:
        reader = csv.DictReader(f)
        headers = reader.fieldnames
        for r in readers := reader:
            uid = r['user_id']
            idx = hash(uid) % num_partitions
            writers[idx].writerow([r[h] for h in headers])
    for t in temps:
        t.close()

    # Reduce phase: aggregate each partition independently
    final = {}
    for t in temps:
        part_sums = defaultdict(float)
        with open(t.name, newline='') as pf:
            reader = csv.DictReader(pf, fieldnames=headers)
            for r in reader:
                part_sums[r['user_id']] += float(r['value'])
        # merge into final (fits because only aggregated keys are kept)
        for k, v in part_sums.items():
            final[k] = final.get(k, 0.0) + v
        os.remove(t.name)
    return final

Key points and reasoning:- In-memory dict approach is simplest and single-pass O(N) time, O(U) space where U = distinct users.- External approach ensures bounded memory by controlling partition size; each partition's distinct users expected ≈ U/num_partitions.- Use hash partitioning so all rows for a user go to same partition.

Complexity:- Time: O(N) total work across phases (plus I/O). Partitioning and reducing are linear passes.- Space: In-memory approach O(U). External approach O(max_partition_size_in_memory) ≈ O(U/num_partitions).

Edge cases:- Malformed rows, non-numeric values → validate/skip with logging.- Choosing num_partitions: trade-off between number of temp files and per-partition memory.- If user_id distribution is highly skewed, use a second-level spill (e.g., further partition heavy keys) or sample to choose N.

Alternative: Use mmap or sqlite3 (standard library) to store partial aggregates on disk if hashing files is inconvenient.

Unlock Full Question Bank

Get access to hundreds of Python Coding and Data Structures interview questions and detailed answers.

Join thousands of developers preparing for their dream job.