Python Programming Questions

Requirements: per-user 7-day rolling sum by event_time, scalable.

Idiomatic, vectorized Pandas (vectorized meaning the loop over rows runs as fast, compiled bulk code instead of a slow Python-level for loop, one row at a time):

import pandas as pd

df = pd.DataFrame({
    "user_id": ["u1", "u1", "u1", "u1", "u2"],
    "event_time": pd.to_datetime([
        "2024-01-01", "2024-01-03", "2024-01-05", "2024-01-09", "2024-01-02"
    ]),
    "value": [10, 20, 30, 40, 5],
})

# ensure datetime and sort
df['event_time'] = pd.to_datetime(df['event_time'])
df = df.sort_values(['user_id', 'event_time'])

# set index for time-based rolling and compute 7-day sum per user
result = (
    df.set_index('event_time')
      .groupby('user_id')['value']
      .rolling('7D')
      .sum()
      .reset_index(name='rolling_7d_sum')
)

Worked example, verified with pandas on CPython 3.12:

df = pd.DataFrame({
    "user_id": ["u1", "u1", "u1", "u1", "u2"],
    "event_time": pd.to_datetime([
        "2024-01-01", "2024-01-03", "2024-01-05", "2024-01-09", "2024-01-02"
    ]),
    "value": [10, 20, 30, 40, 5],
})
print(result)  # (built by running df through the code above)

Output:

  user_id event_time  rolling_7d_sum
0      u1 2024-01-01            10.0
1      u1 2024-01-03            30.0
2      u1 2024-01-05            60.0
3      u1 2024-01-09            90.0
4      u2 2024-01-02             5.0

Tracing u1's rows: Jan 1 has no prior rows in its trailing 7-day window, so the sum is just its own value, 10. Jan 3's window (Jan 3 back to Dec 27) includes Jan 1 and Jan 3, 10 + 20 = 30. Jan 5's window (back to Dec 29) includes Jan 1, 3, and 5, 10 + 20 + 30 = 60. Jan 9's window (back to Jan 2) EXCLUDES Jan 1, which is 8 days earlier, outside the 7-day cutoff, so it sums only Jan 3, 5, and 9: 20 + 30 + 40 = 90. This is exactly what makes it a time-based, not row-count-based, window: the number of prior rows included varies depending on how many actually fall within the trailing 7 real days, not a fixed count of the last N rows.

Notes: Uses time-based rolling with groupby which is vectorized and memory-efficient for large data. For millions of rows, ensure event_time is datetime64, operate on chunked parquet files (parquet: a compressed, columnar file format for tabular data, well suited to reading only the columns and row-groups a job actually needs) if memory constrained, and consider Dask or PySpark for distributed scaling once the data genuinely no longer fits on one machine: Dask mirrors the Pandas API but splits data into partitions and runs the same groupby/rolling-style operations across them in parallel; PySpark is a different, JVM-backed distributed engine with its own (similar but not identical) DataFrame API. Plain, chunked Pandas is enough for anything that still fits on a single machine's memory; reach for Dask/PySpark only once it genuinely does not.

Goal: compare memory and wall-clock of three join methods on two large disk-resident tables.

A concrete illustrative scale, to make the deliverable tangible (placeholder numbers for the experiment's INPUT sizes, not a claimed result; the actual measured numbers only come from running it): joining a 5,000,000-row table to a 50,000,000-row table on a shared user_id column, on a machine with 8 GB of RAM available to the process. That size difference (10x) is deliberately chosen so a hash join (build an in-memory lookup table keyed by the join column from the SMALLER side, then scan the larger side and look up each row's match against it, rather than one built from the bigger side) has an obvious "small side" to build its lookup table from.

Metrics:

• Peak RAM (process) via psutil
• Total wall-clock and CPU time
• I/O throughput and disk usage
• Return correctness (row counts, spot checksums)

Fairness:

• Same machine, same storage medium, same join keys/indexes
• Fixed dataset snapshots and repeat runs per method
• Use warm/cold cache measurements (warm cache: the operating system already has the file's data sitting in memory from a previous read, so the next read is fast; cold cache: the OS has to actually go to disk, which is slower and closer to a fresh, worst-case run; measuring both matters because a method that looks fast on a warm cache can look very different the first time it touches a cold file)

Experiment plan:

• Prepare synthetic datasets with controlled sizes and cardinalities, using the illustrative 5M/50M scale above (or whatever scale matches the real workload) as a concrete starting point
• Method (1): pandas: read table A in chunks, build index on join key for B or stream smaller table and use dict-based hash join (build an in-memory Python dict keyed by the join column from the smaller table, then scan the larger table row by row and look each row's key up in that dict, an $O(n + m)$ approach instead of an $O(n \cdot m)$ nested comparison)
• Method (2): SQLite: load both tables into temp DB, create indexes, run JOIN
• Method (3): Dask (a library that mirrors the Pandas API but splits data into partitions and runs the same operations, including merge/join, across them in parallel, so a join too large for one machine's memory can still complete by processing partitions and spilling intermediate results to disk when needed): persist datasets on disk, execute dask.dataframe.merge with appropriate partitions

Presentation:

• Plot peak memory vs runtime per method (error bars from repeats)
• Table with throughput, disk I/O, and CPU
• Discuss trade-offs: simplicity, indexing cost, parallelism, disk spill behavior (disk spill: what happens when an operation's intermediate data no longer fits in RAM and the engine starts writing partial results to disk to keep going instead of crashing with an out-of-memory error, trading speed for the ability to finish at all)

Success criteria: lowest peak memory with acceptable runtime; reproducible scripts + configs

Plan: do out-of-core aggregation (out-of-core means: process data too big to fit in memory by working on it piece by piece, rather than loading the whole 50 GB file at once) by streaming CSV in chunks or using Dask. Maintain per-key aggregates (sum, count) and compute mean = sum/count.

Chunked Pandas approach:

from collections import defaultdict
import pandas as pd
sums = defaultdict(float)
counts = defaultdict(int)
for chunk in pd.read_csv('big.csv', chunksize=10_000_00):
    grp = chunk.groupby('key')['value'].agg(['sum','count'])
    for k, row in grp.iterrows():
        sums[k] += row['sum']; counts[k] += row['count']
# finalize
result = {k: (sums[k], counts[k], sums[k]/counts[k]) for k in sums}

Worked example (traced on a tiny 6-row file so the shapes are concrete)
Suppose big.csv held just:

key,value
a,10
b,20
a,30
c,5
b,15
a,50

Running the chunked-Pandas code above against this file (even split across multiple small chunks, since each chunk's partial sum/count is added into the running totals rather than overwriting them) produces:

print(result)
# {'a': (90.0, 3, 30.0), 'b': (35.0, 2, 17.5), 'c': (5.0, 1, 5.0)}

Each tuple is (sum, count, mean): key a appears three times with values 10, 30, 50 (sum 90, mean 30.0), key b twice with 20 and 15 (sum 35, mean 17.5), and key c once with 5 (sum 5, mean 5.0). Because the loop only ever adds each chunk's partial sum/count into the running totals, splitting this same 6-row file into 2 chunks of 3 rows each, or processing it as one chunk, produces the identical result dict; only how many times the loop body runs changes, not the answer.

Dask approach: Dask is a library that mirrors the Pandas API (dask.dataframe looks and behaves like pandas.DataFrame) but splits the data into partitions and runs the same operations across them in parallel, so the .groupby().agg() code you already know scales past what fits in one machine's RAM. Use dask.dataframe.read_csv and ddf.groupby('key').agg({'value':['sum','count']}).compute(): Dask handles partitioning and parallelism.

Trade-offs:

• Pure chunking: simple, low memory, single-threaded unless parallelized; needs dictionary sized by number of unique keys.
• Dask: parallel, scales across cores/machines, but adds scheduler overhead and cluster configuration.

Notes: If unique keys are huge, consider external grouping: this is a rare fallback, needed only once the number of distinct keys is itself too large for the sums/counts dictionaries to fit in memory (a separate problem from the original 50 GB of rows, which the chunking above already handles). Two concrete versions of it: sorting by key on disk (writing all rows out sorted so equal keys become adjacent, letting you aggregate in one streaming pass with no dict at all), or using a local key-hash partitioning to spill to disk (hashing each key to decide which of several on-disk partition files it belongs to, so each partition file can later be aggregated on its own, fully in memory, one at a time).

1) Multiprocessing (multiprocessing module)

• Benefit: sidesteps GIL (Global Interpreter Lock: a lock inside CPython that only lets one thread execute Python bytecode at a time, which is why adding more threads alone does not speed up pure-Python CPU-bound work) by using multiple processes; good for CPU-bound tasks that can be partitioned.
• Limitations: IPC and data serialization overhead, higher memory usage per process.
• When: embarrassingly parallel workloads (map-style), batch processing across cores.

2) C-accelerated libraries (NumPy / vectorization)

• Benefit: move heavy numeric loops into C (SIMD, Single Instruction Multiple Data, a CPU feature applying one operation to many values in one step; contiguous memory), drastically faster for array math.
• Limitations: requires expressing work in array form; not helpful for complex control flow.
• When: numerical computations over arrays, matrix ops.

3) Just-in-time / ahead-of-time compilation (Numba or Cython)

• Numba: JIT-compiles (just-in-time compiles: instead of compiling ahead of time before the program ever runs, the function is compiled straight to machine code the first time it is actually called) Python functions to machine code with little code changes; excellent for loops on numeric data.
  • Benefit: low development cost, big speedups.
  • Limitation: not all Python features supported; first-call compilation overhead.
• Cython: compile Python to C for maximum control and speed; can produce greatest gains but requires typing and build step.
  • When: algorithmic hotspots that remain after vectorization or need fine-grained control.

A concrete contrast (real values, not timing, since exact speed depends on hardware and is not something to hardcode here):

import numpy as np

def squares_loop(n):
    return [i * i for i in range(n)]

def squares_vectorized(n):
    return np.arange(n) ** 2

print(squares_loop(5))
# [0, 1, 4, 9, 16]
print(squares_vectorized(5))
# [ 0  1  4  9 16]

Both produce the identical five values; squares_loop pays Python's per-element interpreter overhead five separate times, while squares_vectorized dispatches once into a single compiled C loop, the same mechanism that makes strategy 2 (vectorization) and strategy 3 (Numba/Cython compilation) both faster than a plain interpreted loop: removing per-element interpreter overhead, not changing the underlying arithmetic.

Choose based on workload: use vectorization first, then Numba for loop-heavy numeric code, and multiprocessing when parallelism across cores is needed and data can be partitioned.

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.