Python Programming Questions

Comparison:

• Method A (single-threaded generator): simple, low overhead, minimal memory, deterministic ordering, easy error handling; limited by single-core CPU and Python parsing speed.
• Method B (multi-process byte-range split): parallel CPU usage, faster parse throughput on multi-core, but more complex (must align splits to line boundaries), higher peak memory (multiple workers), and increased I/O contention.

Method A, concretely (the whole method really is this short, which is part of why it is the default):

import json

def read_jsonl(path):
    with open(path, 'r', encoding='utf-8') as f:
        for line in f:
            yield json.loads(line)

Method B, concretely, with real byte offsets so 'split misalignment' can be traced rather than only named: take a tiny 40-byte JSON-lines file, four records of exactly 10 bytes each, {"id": 1}\n{"id": 2}\n{"id": 3}\n{"id": 4}\n (bytes 0-9 are line 1, 10-19 line 2, 20-29 line 3, 30-39 line 4). Splitting this file evenly at byte 20 happens to land exactly on a line boundary, so nothing goes wrong there. Splitting unevenly, worker 0 gets raw bytes [0, 15), worker 1 gets [15, 40), lands worker 1's start offset at byte 15, verified to be the : character in the middle of line 2, not the start of any record:

def worker_range(path, start, end):
    with open(path, 'rb') as f:
        f.seek(start)
        if start != 0:
            f.readline()          # discard the partial line landed on, if any
        while f.tell() < end:
            line = f.readline()
            if not line:
                break
            yield json.loads(line)

Worker 1 seeks to byte 15 (mid-line-2), then its f.readline() reads and discards the remainder of line 2 (bytes 15-19, ': 2}\n'), landing the cursor exactly at byte 20, the true start of line 3; from there it reads whole lines normally. The other half of line 2 (bytes 10-15) was already skipped by worker 0, which stopped consuming at byte 20 without seeing it either, so line 2 as a whole is silently dropped. This is exactly the alignment bug: each worker must seek to (and discard up to) the next newline after its own raw start offset, or a record straddling a boundary is lost or corrupted.

When B is preferable:

• CPU-bound parsing (complex JSON) and available cores/IO bandwidth
• File stored on fast SSD or networked storage that supports concurrent reads

Pitfalls & failure modes:

• Split misalignment: worker starts mid-line: must seek to next newline
• Memory pressure: many workers each allocate buffers; can OOM
• Ordering: results may be out-of-order; need merging if order matters
• Partial line encoding issues (multibyte UTF-8 boundaries: UTF-8 is a variable-width encoding where one character can take anywhere from 1 to 4 bytes, so a byte-offset split chosen without regard to character boundaries can land in the middle of a multi-byte character, not just in the middle of a line; a correct implementation aligns on a newline byte, which is always a single, unambiguous ASCII byte and never appears as a continuation byte inside a multi-byte UTF-8 character, rather than aligning on an arbitrary byte count)
• Error recovery complexity: a worker crash loses its chunk

Guidelines: choose worker count to match CPU/IO balance, implement safe split alignment, stream results via queues, and fallback to single-threaded if resources constrained.

Approach
Run import in a subprocess to enforce timeout and capture import errors/syntax errors without raising in caller.

Implementation:

import subprocess, sys, json, tempfile

def safe_import(module_name, timeout=5):
    code = f"import importlib, json\ntry:\n m=importlib.import_module('{module_name}')\n print('OK')\nexcept Exception as e:\n print('ERR:'+type(e).__name__+':'+str(e))\n"
    proc = subprocess.run([sys.executable, '-c', code], capture_output=True, text=True, timeout=timeout)
    out = proc.stdout.strip()
    if out.startswith('OK'):
        return __import__(module_name), None
    if out.startswith('ERR:'):
        return None, out[4:]
    return None, 'Unknown error'

Why the module gets imported twice
The subprocess step and the __import__ step are doing two different jobs, not one job done redundantly. The subprocess runs in a completely separate operating-system process, with its own memory, so whatever module object it constructs cannot be handed back to the caller at all, a subprocess and its parent only communicate through the captured stdout text, 'OK' or 'ERR:...', never through live Python objects. So the subprocess's only purpose is to safely test-drive the import (catching a hang via timeout, or a crash, or a SyntaxError from a broken package) without risking the caller's own process. Once that test comes back 'OK', the code still needs an actual, usable module object inside the CALLER's process, which is exactly what the second call, __import__(module_name), provides; it re-runs the import for real, now that it is known to be safe, in the process that actually needs the result.

Worked example, verified on CPython 3.12:

mod, err = safe_import('json')
print(mod is not None, err)
# True None

mod2, err2 = safe_import('this_module_does_not_exist')
print(mod2, err2)
# None ModuleNotFoundError:No module named 'this_module_does_not_exist'

For a real, importable module, safe_import returns a usable module object and None for the error; for a nonexistent one, it returns None and the exact type/message the subprocess's except Exception as e branch captured, with nothing raised in the caller.

Notes

• Using subprocess prevents a broken package from crashing the caller and enforces timeout.
• For heavy imports this adds overhead; use caching for repeated imports.
• Captures ImportError, SyntaxError, or runtime exceptions during module import.

Goal: return k most frequent items from a possibly streaming iterable using O(k) extra memory where possible.

Approach: Use Misra-Gries (frequent algorithm) for streaming approximate top-k with O(k) memory; for exact top-k when feasible, use counting with a hashmap and a min-heap of size k (memory O(u) where u unique). For very large universes, prefer Misra-Gries. This is a genuinely advanced, fairly obscure streaming algorithm, most engineers have never needed it; skip straight to "hashmap + heap, exact counts" below unless the interviewer specifically probes for a single-pass, bounded-memory answer.

The core intuition, in plain language, before any code: Misra-Gries tracks at most k candidate items with counts. When a new, not-yet-tracked item arrives and there is no room left (k candidates already exist), instead of dropping the new item and keeping the old ones untouched, EVERY existing candidate's counter is decremented by one, as if the new item had "cancelled out" one occurrence of each current candidate, and the new item itself is not added. Any item that truly appears more than roughly n/k times (n being the total stream length) cannot be fully cancelled away by this process, no matter how the decrements land, so it is guaranteed to still be a candidate at the end; the guarantee is one-sided, though, some less-frequent items may also survive as candidates (false positives), which is exactly why the code takes a second pass over the real data to compute true counts for whatever candidates survived, rather than trusting the approximate counters directly.

Misra-Gries implementation (approximate, deterministic guarantees):

from collections import defaultdict

def top_k_frequent(stream, k):
    if k < 1:
        return []
    counters = {}
    for x in stream:
        if x in counters:
            counters[x] += 1
        elif len(counters) < k:
            counters[x] = 1
        else:
            # decrement all
            to_del = []
            for y in list(counters):
                counters[y] -= 1
                if counters[y] == 0:
                    del counters[y]
    # counters are candidates; to get actual counts, re-scan
    true_counts = defaultdict(int)
    for x in stream:
        if x in counters:
            true_counts[x] += 1
    return sorted(true_counts.items(), key=lambda t: -t[1])[:k]

Worked trace, verified on CPython 3.12: an 8-item stream with k=2 (so the true top item, 'a', appears 5 out of 8 times, far more than the other three items, each appearing once):

stream = ['a', 'a', 'b', 'c', 'a', 'd', 'a', 'a']
print(top_k_frequent(stream, 2))
# [('a', 5), ('d', 1)]

Tracing counters step by step: a (candidate, table has room) -> {'a': 1}. Second a (already tracked, increment) -> {'a': 2}. b (room) -> {'a': 2, 'b': 1}. c arrives with the table full (2 candidates already, k=2): every existing candidate is decremented, a drops to 1, b drops to 0 and is deleted, and c itself is never added -> {'a': 1}. Third a (tracked, increment) -> {'a': 2}. d (room again, since b was just evicted) -> {'a': 2, 'd': 1}. Fourth and fifth a (tracked, increment twice) -> {'a': 4, 'd': 1}. Final candidates: a and d. The second pass then counts each candidate's REAL frequency across the whole stream (a: 5, d: 1) and returns them sorted by that real count.

This trace also shows the false-positive behavior honestly: b and c both had a true count of 1, identical to d's true count of 1, but only d happened to survive as a candidate, purely because of when it arrived relative to the decrement-all events. The algorithm's real guarantee is only about the dominant item: a, at 5 out of 8 occurrences, appears far more often than roughly n/k = 4, and it survives every decrement round intact enough to still be tracked at the end. Nothing is guaranteed about which of the equally-infrequent items happen to survive alongside it, which is precisely why a second, exact pass over the real data is required before trusting any reported count, the approximate candidate set is a correct superset guarantee for truly frequent items, not a precise answer on its own.

Notes: The algorithm uses O(k) memory and one or two passes (approximate single-pass; exact requires second pass over data or storing counts). Choose Misra-Gries for streaming huge data; use Counter+heap for exact counts when unique items fit memory.

Possible Python/library causes

• dict ordering: dict insertion order IS deterministic in CPython 3.7+ (this is a language guarantee, not an implementation quirk), but the actual non-determinism seen in production comes from something upstream feeding the dict in a different order between runs, for example a filesystem glob() returning filenames in a different order, or multiple workers finishing and inserting their results in whatever order they happen to complete, not the dict itself misbehaving.
• set iteration: unordered, causes non-deterministic traversal.
• floating-point reductions (sum/mean) non-associative ordering leads to tiny differences.
• parallelism: thread/process scheduling, race conditions, non-deterministic map ordering.
• use of RNG without fixed seeds (random, numpy.random, torch).
• reliance on OS-level nondeterminism (filesystem glob order, multiprocessing.Queue timing).

Two of these causes, made concrete, verified on CPython 3.12:

Non-associative floating-point reduction, a naive accumulator loop genuinely disagreeing with itself depending on order (built-in sum() on recent CPython uses a more accurate internal algorithm and happens to avoid this specific example, which is exactly why the naive loop below, closer to what a hand-rolled or per-worker partial-sum reduction actually does, is the more honest demonstration):

def manual_sum(values):
    total = 0.0
    for v in values:
        total += v
    return total

vals = [0.1, 0.2, 0.3]
print(manual_sum(vals))
# 0.6000000000000001
print(manual_sum(list(reversed(vals))))
# 0.6

import math
print(math.fsum(vals) == math.fsum(list(reversed(vals))))
# True

The same three floats, summed in two different orders by a plain accumulator, produce two DIFFERENT bit patterns, 0.6000000000000001 versus exactly 0.6, purely from the order of addition (this is exactly what happens when parallel workers each compute a partial sum and the main process combines them in whatever order they finish); math.fsum (a compensated-summation algorithm that tracks and corrects rounding error as it goes) gives the identical, order-independent result either way.

Unseeded RNG producing different output on repeated runs, and the fix:

import random

random.seed(42)
print(random.random())
# 0.6394267984578837
random.seed(42)
print(random.random())
# 0.6394267984578837 -- identical, since the seed was reset

Calling random.seed(42) again before the second random.random() call reproduces the exact same float; omitting the seed (or seeding from the current time, which is random's default) would make this genuinely different on every run, which is exactly the failure mode the fix below addresses.

Concrete fixes

• Sort keys/lists before iteration (e.g., sorted(dict.items())).
• Use deterministic data structures: collections.OrderedDict where explicit order required.
• Fix RNG seeds at pipeline start for all libs (random.seed, np.random.default_rng with seed passed).
• Use stable reduction algorithms: math.fsum or pairwise summation (numpy.sum with dtype and stable algorithm libs).
• Avoid unordered set for grouping; use sorted containers or deterministic grouping keys.
• Control parallelism: fix number of workers, use deterministic partitioning, or run single-threaded for critical steps.
• Pin library versions and document environment (requirements, Docker).

These steps produce reproducible outputs suitable for production ETL runs and debugging.

How I’d debug it

I’d separate the problem into three buckets: I/O, string work, and memory growth.

• Measure first: run the job with cProfile or simple timers around file open/read, parsing, and final aggregation. If file read time dominates, it is I/O. If CPU time is in split, replace, regex, or repeated concatenation, it is string processing. If runtime is okay but resident memory in RAM keeps climbing, use tracemalloc to see what objects are retained.
• Look for repeated file reads: if the same file is opened more than once per run, cache the parsed content in a single pass.
• Watch for holding too much data: avoid building a giant list if you only need counts or totals.

Worked example, made concrete with a small stand-in file (1,000 lines) and real, counted line-visits rather than a timing claim, verified on CPython 3.12:

sample_log = "\n".join(f"line-{i}" for i in range(1000))   # stand-in for the real 200 MB log

visits_three_pass = 0
def summary_errors(text):
    global visits_three_pass
    for line in text.splitlines():
        visits_three_pass += 1

def summary_warnings(text):
    global visits_three_pass
    for line in text.splitlines():
        visits_three_pass += 1

def summary_totals(text):
    global visits_three_pass
    for line in text.splitlines():
        visits_three_pass += 1

summary_errors(sample_log)
summary_warnings(sample_log)
summary_totals(sample_log)
print(visits_three_pass)
# 3000

visits_one_pass = 0
for line in sample_log.splitlines():
    visits_one_pass += 1   # all three counters would be updated right here, in this one visit
print(visits_one_pass)
# 1000

The three-separate-reads version visits every line 3000 times total (1,000 lines x 3 passes); the single-pass version visits each line exactly once, 1,000 times total, and updates all three counters during that same visit. This ratio (here exactly 3x, matching the number of summaries) is what actually shrinks, both the I/O (the file itself would also only be opened and read once, instead of three times) and the CPU work of iterating the lines; it is a real, derivable count rather than a machine-specific timing number, and it scales the same way whether the log is 1,000 lines or 200 MB. I would change it to one pass that parses each line once and updates all three counters. That usually fixes both slow runtime and memory pressure.

First change: stream the file line by line and keep only the aggregate data you truly need. That is the highest-impact fix before micro-optimizing string operations.