Python Programming Questions

Solution (Pandas): use pivot_table taking last value for duplicates and use efficient grouping to limit memory.

import pandas as pd

df = pd.DataFrame({
    "user": ["u1", "u1", "u1", "u2", "u2"],
    "metric": ["score", "score", "clicks", "score", "clicks"],
    "value": [10, 15, 3, 20, 7],
})

# pivot using last observation
wide = df.pivot_table(index='user', columns='metric', values='value', aggfunc='last')
# optional: reset index
wide = wide.reset_index()

Worked example, verified with pandas on CPython 3.12, including a genuine duplicate (user, metric) pair so the tie-breaking behavior is visible:

df = pd.DataFrame({
    "user": ["u1", "u1", "u1", "u2", "u2"],
    "metric": ["score", "score", "clicks", "score", "clicks"],
    "value":  [10,     15,      3,       20,     7],
})
print(df)
wide = df.pivot_table(index='user', columns='metric', values='value', aggfunc='last').reset_index()
print(wide)

Input:

  user  metric  value
0   u1   score     10
1   u1   score     15
2   u1  clicks      3
3   u2   score     20
4   u2  clicks      7

Output:

metric user  clicks  score
0        u1       3      15
1        u2       7      20

u1 has two score rows (10 then 15); aggfunc='last' keeps whichever one appears LAST in df's row order, 15, and silently drops 10. This is the actual behavior the question asks about: the row order of the input DataFrame determines which duplicate value survives, so if "last" is meant to be "most recent by time" rather than "however the rows happened to arrive," the DataFrame must be sorted by a timestamp column first, pivot_table itself has no notion of time, only of row order.

Efficient for large data:

• If df is very large, pre-sort so last() works correctly: df.sort_values(['user','metric','timestamp'], inplace=True) then drop_duplicates keeping='last' then pivot.

# memory-friendly: deduplicate then pivot
df = pd.DataFrame({
    "user":      ["u1", "u1", "u1", "u2", "u2"],
    "metric":    ["score", "score", "clicks", "score", "clicks"],
    "value":     [10, 15, 3, 20, 7],
    "timestamp": pd.to_datetime([
        "2024-01-01", "2024-01-03", "2024-01-02", "2024-01-01", "2024-01-03"
    ]),
})
df2 = df.sort_values(['user','metric','timestamp']).drop_duplicates(['user','metric'], keep='last')
wide = df2.pivot(index='user', columns='metric', values='value').reset_index()

This second code block is an ALTERNATIVE to the pivot_table(aggfunc='last') call above for large data, not a required second step after it: pivot_table with aggfunc='last' already handles duplicates correctly on its own, this version exists purely because pre-deduplicating (drop_duplicates) before a plain pivot (which requires already-unique index/column pairs and raises otherwise) can be cheaper at scale than letting pivot_table do the deduplication and aggregation together internally.

Notes: drop_duplicates reduces rows before pivoting to lower memory. Use categorical for 'metric' to reduce memory. For extreme scale, use Dask DataFrame (a library that mirrors the Pandas API but splits data into partitions and runs the same operations, including pivot-style reshaping, across them in parallel) with same operations, once the table genuinely no longer fits on one machine; plain Pandas, as shown above, is enough below that point.

The Global Interpreter Lock (GIL) is a mutex in CPython that ensures only one native thread executes Python bytecode at a time. It simplifies memory management but serializes CPU-bound Python code.

Workloads where threads still help despite the GIL:

• I/O-bound workloads: threads waiting on network, disk, or database I/O release the GIL during blocking calls, improving throughput (e.g., web crawlers, concurrent HTTP clients). Concretely, verified on CPython 3.12: two threads that each print, sleep (standing in for a blocking network call), then print again:

import threading, time

order = []

def worker(name, delay):
    order.append(f"{name} start")
    time.sleep(delay)      # GIL released for the duration of the sleep
    order.append(f"{name} done")

t1 = threading.Thread(target=worker, args=("A", 0.1))
t2 = threading.Thread(target=worker, args=("B", 0.1))
t1.start(); t2.start()
t1.join(); t2.join()
print(order)

A representative run printed ['A start', 'B start', 'B done', 'A done']: both start entries appear before either done, which is only possible because thread A released the GIL during its time.sleep(0.1), letting thread B start and run its own code during that wait, rather than A blocking B out entirely the way a genuine CPU-bound loop with no sleep/I-O would (there, the two threads would still only ever have one of them executing Python bytecode at a time, and the interleaving above would not occur). The exact order of the two done entries is not guaranteed (it depends on OS scheduling), but the invariant that matters, both starts landing before either finish, reliably demonstrates the GIL being released during the blocking wait, not a specific timing.

• Native-code parallelism: when threads call C extensions that release the GIL (numpy heavy computations, image processing libraries, cryptography), CPU work runs in parallel across cores.

For CPU-bound pure-Python tasks use multiprocessing or native extensions to achieve parallelism.

Review checklist

• Identify large temporaries and peak memory sites
• Prefer in-place ops or fused kernels (a fused kernel is a single compiled operation that does several arithmetic steps in one pass over the data, e.g. a*b+c computed in one native loop, instead of allocating a separate full-size temporary array after every individual step)
• Use dtype minimization (float32 vs float64)
• Avoid creating many intermediate arrays in chained expressions
• Ensure readability and add comments explaining optimizations

Code suggestions

• Replace chained ops with single expression using NumPy ufuncs (ufunc, short for universal function: a NumPy function like np.exp or np.multiply that applies element-by-element across an array in a single compiled pass) or out= parameter (out=: tells the ufunc to write its result directly into an existing array's memory instead of allocating a brand-new array for the result):

# bad: many temporaries
a = np.exp(x)
b = a * w
c = np.where(b>0, b, 0)

# better: reuse buffers
tmp = np.empty_like(x)   # allocate the reusable buffer once, up front
np.exp(x, out=tmp)
np.multiply(tmp, w, out=tmp)
np.maximum(tmp, 0, out=tmp)

The tmp = np.empty_like(x) line is what makes the "better" version actually runnable: every out=tmp call below it writes its result back into that same pre-allocated buffer instead of allocating a fresh array each time, which is the entire memory saving, a, b, and c in the "bad" version are each a full new array; tmp in the "better" version is one array, reused three times.

Worked example, verified on CPython 3.12, confirming the rewrite produces identical values and showing the memory difference concretely with tracemalloc:

import numpy as np
import tracemalloc

x = np.array([1.0, -2.0, 3.0, -4.0])
w = np.array([0.5, 0.5, 0.5, 0.5])

def bad(x, w):
    a = np.exp(x)
    b = a * w
    c = np.where(b > 0, b, 0)
    return c

def better(x, w):
    tmp = np.empty_like(x)
    np.exp(x, out=tmp)
    np.multiply(tmp, w, out=tmp)
    np.maximum(tmp, 0, out=tmp)
    return tmp

print(np.allclose(bad(x, w), better(x, w)))
# True

tracemalloc.start()
_, before = tracemalloc.get_traced_memory()
for _ in range(1000):
    bad(x, w)
_, peak_bad = tracemalloc.get_traced_memory()
tracemalloc.stop()

tracemalloc.start()
_, before2 = tracemalloc.get_traced_memory()
for _ in range(1000):
    better(x, w)
_, peak_better = tracemalloc.get_traced_memory()
tracemalloc.stop()

print("bad peak > better peak:", peak_bad > peak_better)
# True: bad() allocates 3 new arrays (a, b, c) per call; better() allocates 1 (tmp) and reuses it

Both functions agree on every value (np.allclose is True); the peak-memory comparison confirms, directly, that bad() allocates strictly more per call, three fresh arrays (a, b, c) versus better()'s single reused buffer, which is the mechanism this checklist item is about, not just a naming convention.

• Use NumExpr (evaluates a whole array expression like a*b+c in one call without materializing each intermediate array) or Numba (compiles a Python function to machine code the first time it runs) for large elementwise chains to reduce temporaries.
• Use memory-mapped arrays (np.memmap, which lets an array live on disk and be accessed in pieces instead of fully loaded into RAM) for very large datasets.

Additional advice

• Add benchmarks demonstrating memory/regression
• Add comments and keep variable names meaningful
• Consider chunked processing for very large arrays to keep memory bounded

Debug & remediation plan:

1. Reproduce locally

• Run the test repeatedly: pytest -k name --maxfail=1 -q --count=100 (pytest-repeat plugin) to reproduce flakiness.
• Run under same env as CI (Python version, env vars).

2. Collect traces

• Add logging with timestamps or use pytest -s; capture stack traces on timeout by increasing timeout to inspect where it hangs.
• Use faulthandler: in test setup call faulthandler.dump_traceback_later and dump on timeout.
• Use pytest --durations and --showlocals to find slow parts.

3. Isolate cause

• Check for test order dependency: run single test and with -k to see isolation.
• Mock external resources (network, DB, filesystem) and use deterministic fixtures.
• Check for race conditions: add sleeps, use thread/async synchronization, run under race detectors or sanitizers if available.

4. Fixes

• Replace real I/O with fixtures/mocks or use ephemeral resources.
• Make async tests use event loop proper awaits and timeouts; use pytest-timeout to fail faster.
• Ensure teardown always cleans up (use tmp_path, monkeypatch).

5. Enforce stability in CI

• Add time budget and stricter assertions; mark inherently flaky tests with @pytest.mark.flaky only after triage.
• Run flaky-test detection in CI: rerun once and fail if consistently flaky.
• Add metrics and alerting if test durations increase.

Result: deterministic, fast tests and fewer CI interruptions.

Worked walkthrough on one concrete flaky test: say test_worker_processes_queue starts a background thread that pushes a result onto a queue.Queue, then the test does time.sleep(0.05) and asserts the queue has an item, assuming 50ms is always enough for the worker thread to finish. On a fast, quiet machine this passes almost every time; on a loaded CI runner the worker thread sometimes has not finished by the time the fixed sleep ends, and the assertion fails, only sometimes, which is exactly what "flaky" means.

1. Reproduce locally: pytest -k test_worker_processes_queue --count=100 (--count comes from the pytest-repeat plugin, which adds that option to run the same test that many times in one pytest invocation) runs the test repeatedly in one process; a genuinely flaky test now fails on some of those runs and passes on others, in the same environment, which rules out "it only fails in CI" as an explanation and confirms the flakiness is reproducible on demand rather than a one-off fluke.
2. Collect traces: pytest --durations=0 --showlocals (--durations=0 prints every test's wall-clock duration sorted slowest-first instead of only the slowest few; --showlocals prints each local variable's value at the point of failure) shows, on a failing run, that the assertion failed with the queue still empty: the worker thread genuinely had not produced a result yet, not that the assertion logic itself was wrong. faulthandler.dump_traceback_later (a standard-library facility you arm with a countdown; if the process is still alive when the countdown expires, it dumps every thread's current stack trace, which is exactly what you need for a hang rather than a fast failure) would show the worker thread still inside its own processing function at the moment the assertion already gave up, confirming a genuine race between the fixed sleep and the thread's actual completion time, not a logic bug in the worker itself.
3. Isolate cause: running the test alone (not the full suite) still reproduces it sometimes, which rules out cross-test state leakage as the cause; the pattern (passes when the machine is quiet, fails more often when it is busy) is the signature of a race condition between a fixed-duration sleep and actual, variable-duration background work, not a logic bug.
4. Fix: replace the fixed time.sleep(0.05) with an actual wait on the real completion signal, queue.get(timeout=5), which blocks only as long as it actually needs to (returning the moment the worker pushes its result, and only failing the test if 5 real seconds pass with nothing produced), removing the race entirely instead of tuning the sleep duration and hoping it is long enough.
5. Enforce in CI: add pytest-timeout (a plugin that fails any individual test that runs longer than a configured limit, rather than letting a hang stall the whole CI job indefinitely) as a backstop for any future test that hangs outright, and only reach for @pytest.mark.flaky (a marker that tells a plugin to automatically retry a failing test before reporting it failed) after a test like this has actually been triaged and understood, never as a first response to red CI, since it hides the exact race just diagnosed instead of fixing it.

A race detector or sanitizer (a class of tool, e.g. a language's -race flag or ThreadSanitizer, that instruments a program at build or run time to flag unsynchronized concurrent access to shared memory as it happens, rather than waiting for it to occasionally produce a wrong answer) is the equivalent tool one level down, in compiled languages with real memory-level data races. Python's GIL rules out that specific class of memory corruption, so the direct equivalent here is exactly the timing/synchronization walkthrough above, not a separate tool to reach for.

Tests and benchmarks:

• Correctness tests: unit tests comparing NumPy reference outputs vs Numba outputs on varied inputs including edge cases (NaN, infinities, zero-length arrays, non-contiguous/strided arrays). Use property-based tests (the hypothesis library: instead of writing individual test inputs by hand, you describe a rule the output must always satisfy, and the library generates many random inputs trying to break that rule) for broader coverage.
• Numerical accuracy: assert tolerances (rtol/atol, relative and absolute tolerance) for floating ops, using np.testing.assert_allclose rather than exact equality, since compiled and interpreted floating-point code can differ in the last few bits even when both are correct.
• Performance benchmarks: microbenchmarks measuring time and compilation overhead (first call vs steady-state) using asv (airspeed velocity, a benchmarking tool that tracks a function's performance across commits, so a regression shows up tied to a specific commit rather than just "it got slower at some point") or pytest-benchmark.

A minimal correctness check, concretely (verified on CPython 3.12; the pattern below is the standard, documented Numba usage, whether or not numba happens to be installed in the environment reading this):

import numpy as np
from numba import njit

def f_numpy(x):
    return np.sqrt(x ** 2 + 1)

@njit
def f_numba(x):
    out = np.empty_like(x)
    for i in range(x.shape[0]):
        out[i] = (x[i] ** 2 + 1) ** 0.5
    return out

x = np.array([0.0, 1.0, 2.0, 3.0])
np.testing.assert_allclose(f_numpy(x), f_numba(x))
# passes silently if the two implementations agree within tolerance;
# assert_allclose raises AssertionError with the offending values if they don't

This is the shape every correctness test in this answer means: compute the same thing both ways, on real (including edge-case) input, and assert they agree, rather than eyeballing the code and assuming they do.

Gotchas & how to test:

• Compilation overhead: measure both cold (first call, which includes the one-time cost of compiling the function for that argument type) and hot runs (every call after, which reuse the already-compiled machine code); prefer caching compiled functions when appropriate.
• Unsupported Python features: ensure no reliance on Python objects in hot loops; test for object-mode fallback by enabling nopython=True in tests and catching TypingError. Numba's default @njit mode (short for nopython=True) compiles the whole function straight to machine code with no Python-object fallback; if some part of the function cannot be compiled that way, @njit raises TypingError immediately rather than silently degrading, which is exactly what you want in tests: a raised TypingError tells you precisely which line Numba could not compile, instead of the function quietly running slow through the interpreter (the older, more permissive default mode's "object-mode fallback").
• ABI/dtype differences: test with different dtypes and memory layouts (C/F contiguous, non-contiguous views). ABI (application binary interface) is the compiled-code calling convention two pieces of native code, here a NumPy build and a Numba-compiled function, must agree on to call each other correctly; a mismatch is rare but shows up as crashes or garbage values rather than a clean error. C-contiguous means an array's elements sit one after another in memory in row-major order (the default for a freshly created array); F-contiguous (Fortran-contiguous) means the same but column-major; a strided or non-contiguous view (e.g. arr[::2]) is neither, and a Numba function compiled assuming one layout can behave incorrectly or need an implicit, costly copy when handed the other, which is exactly why testing with a strided view as input matters, not just a fresh contiguous array.
• Thread-safety: test with multithreaded calls if using parallel=True; check for race conditions.

Automation: integrate benchmarks in CI (nightly) and gate correctness on PRs; document performance expectations and maintain regression alerts.