Python Programming Questions

Difference & when to prefer

• List comprehension: produces a list eagerly; concise and fast for moderate-sized results when you need random access or repeated iteration.
• Generator expression: lazy iteration, low memory; prefer for large streams or pipeline processing.
• map/filter: functional style; can be slightly faster in some cases and composes well with other functions or builtins; returns iterator in Py3.

Examples (square 0..9):
List comprehension:

squares = [x*x for x in range(10)]
print(squares)
# [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

Generator expression:

squares_gen = (x*x for x in range(10))
print(list(squares_gen))
# [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

map/filter:

squares_map = list(map(lambda x: x*x, range(10)))
print(squares_map)
# [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

All three print the identical list, confirming the three forms are interchangeable here; only their evaluation style (eager list, lazy generator, eager map/filter) differs.

Situation: Overnight job reads a directory of daily status files. Some may be empty, malformed, or missing.

Task: Keep valid data moving, but tell the operator exactly what failed.

Action: I would loop over the expected file list and handle errors per file, not around the whole batch.

• If a file is missing, record missing and continue.
• If a file is empty, record empty and continue.
• If a line is malformed, skip just that file or just that line depending on the rule, and capture the filename and line number.
• Keep running totals in memory, plus a failed_files list for the final report.

Result: The job still produces a useful summary from good files, while the operator gets a clean failure section like status_2025-07-03.txt: malformed line 42 and status_2025-07-04.txt: missing. That is better than a hard stop because it keeps the daily workflow moving.

A small worked example, verified on CPython 3.12, showing the actual loop and its exact printed summary rather than only describing the shape:

def build_report(expected_files, file_contents):
    report = {"processed": 0, "total_value": 0, "failed": []}
    for name in expected_files:
        if name not in file_contents:
            report["failed"].append(f"{name}: missing")
            continue
        content = file_contents[name]
        if content.strip() == "":
            report["failed"].append(f"{name}: empty")
            continue
        try:
            value = int(content.strip())
        except ValueError:
            report["failed"].append(f"{name}: malformed line")
            continue
        report["processed"] += 1
        report["total_value"] += value
    return report

expected = [f"status_2025-07-{i:02d}.txt" for i in range(1, 11)]
contents = {name: "5" for name in expected}
contents["status_2025-07-05.txt"] = ""          # empty
del contents["status_2025-07-09.txt"]            # missing

print(build_report(expected, contents))

Output:

{'processed': 8, 'total_value': 40, 'failed': ['status_2025-07-05.txt: empty', 'status_2025-07-09.txt: missing']}

8 files processed (matching the 10 expected minus the 1 empty and 1 missing), each contributing a value of 5, giving total_value = 40; the two problem files are named explicitly in failed with the specific reason, exactly the breakdown a report shown to an operator needs.

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

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).

Approach: stream the file line-by-line using an internal buffer of chunk_size, parse CSV incrementally, yield namedtuples and convert numeric-looking strings to int/float lazily.

Implementation:

import csv
from collections import namedtuple
from io import TextIOBase

def stream_csv(path_or_file, chunk_size=65536):
    f = open(path_or_file, 'r', newline='') if isinstance(path_or_file, str) else path_or_file
    reader = csv.reader(f)
    headers = next(reader)
    Row = namedtuple('Row', [h.replace(' ', '_') for h in headers])
    def conv(s):
        if s == '':
            return None
        try:
            i = int(s)
            return i
        except ValueError:
            try:
                return float(s)
            except ValueError:
                return s
    for raw in reader:
        yield Row(*[conv(field) for field in raw])
    if isinstance(path_or_file, str):
        f.close()

A namedtuple (created above via namedtuple('Row', [...])) is a lightweight, tuple-backed class whose fields you can access by name, row.age, instead of only by position, row[1], which is why it is used here rather than a plain tuple or a dict: it is as compact and fast as a tuple, but each yielded row reads like a normal object with named fields.

Worked example, verified on CPython 3.12:

import io

sample_csv = "name,age,score\nAlice,30,9.5\nBob,,7\n"
rows = list(stream_csv(io.StringIO(sample_csv)))
print(rows)
# [Row(name='Alice', age=30, score=9.5), Row(name='Bob', age=None, score=7)]
print(rows[0].name, rows[1].age)
# Alice None

conv turns '30' into the int 30 and '9.5' into the float 9.5; Bob's empty age field hits the s == '' check first and becomes None rather than being run through int()/float() at all, and rows[0].name shows the named-field access a plain tuple could not offer without remembering which index held which column.

Key points: single-pass, constant memory per-row, converts numbers eagerly per-field. For truly huge fields, use streaming CSV parsers (csv module is already streaming).

Complexity: O(n) time where n is total fields; O(1) extra space per row.

Edge cases: quoted newlines, malformed rows, large field sizes: consider using csv.DictReader and incremental file objects if needed.