Find a Slow CSV Lookup with Python cProfile and Verify the Optimization

A scheduled job finishes, but processing its growing CSV takes longer each week. Standard-library cProfile and pstats can show where repeated work happens. This tutorial finds repeated list scans, changes one lookup container, then checks both output and elapsed time.

It is for beginner-to-intermediate readers reproducing the example locally on Linux. Use CPython 3.10 or newer; no third-party packages are needed. The examples were tested on Linux with CPython 3.11.4. For a job that stops making progress, start with faulthandler thread-stack troubleshooting.

1. Method: locate the cost, then validate

1. Profile the list version against a fixed input.
2. Locate lookup work in is_allowed() and build a set outside the loop.
3. Check that the two CSV files match byte for byte, then time the whole command without profiling.

This example performs existence checks on string IDs. Repeated IDs in the allow-list must not multiply output rows. Repeated input records and their order must survive. The script does not perform a relational join or load every input row into memory.

2. Set up the environment and data

1. Create a practice directory in the same terminal and run subsequent commands there:

python3 --version
python3 -m cProfile --help
mkdir cprofile-csv-demo
cd cprofile-csv-demo

2. Generate 20000 records and 5001 allow-list entries. There are only 5000 distinct allowed IDs, and every input ID occurs twice. Some notes contain Chinese text, commas, and newlines.

python3 - <<'PY_DATA'
import csv
from pathlib import Path

with open("samples.csv", "x", encoding="utf-8", newline="") as file:
    writer = csv.writer(file)
    writer.writerow(["sample_id", "note"])
    for i in range(20000):
        note = f"record {i}"
        if i % 4999 == 0:
            note = f"记录 {i}, item\nsecond line"
        writer.writerow([f"S{i % 10000:05d}", note])

with Path("allowed.txt").open("x", encoding="utf-8") as file:
    for i in range(0, 10000, 2):
        file.write(f"S{i:05d}\n")
    file.write("S00000\n")
print("Created 20000 records and 5001 allow-list entries")
PY_DATA

Data files use exclusive creation mode, x, so an existing file raises an error. Use a new directory to repeat the exercise. Following the official CSV documentation, files are opened with newline="" so the parser handles multiline fields.

3. Save the complete filter

Save this as filter_samples.py:

"""Profile list versus set membership while filtering a demo CSV."""
import argparse
import csv
from pathlib import Path
import sys


def is_allowed(sample_id, allowed_ids):
    return sample_id in allowed_ids


def filter_samples(source, allow_file, output, index):
    allowed_ids = allow_file.read_text(encoding="utf-8").splitlines()
    if index == "set":
        allowed_ids = set(allowed_ids)
    kept = 0
    with source.open(encoding="utf-8-sig", newline="") as incoming:
        reader = csv.reader(incoming, strict=True)
        if next(reader, None) != ["sample_id", "note"]:
            raise ValueError("expected header: sample_id,note")
        with output.open("x", encoding="utf-8", newline="") as outgoing:
            writer = csv.writer(outgoing)
            writer.writerow(["sample_id", "note"])
            for number, row in enumerate(reader, 1):
                if len(row) != 2 or not row[0]:
                    raise ValueError(f"invalid data record {number}")
                if is_allowed(row[0], allowed_ids):
                    writer.writerow(row)
                    kept += 1
    return kept


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("source", type=Path)
    parser.add_argument("allow_file", type=Path)
    parser.add_argument("output", type=Path, help="new output CSV; must not exist")
    parser.add_argument("--index", choices=["list", "set"], default="list")
    args = parser.parse_args()
    try:
        kept = filter_samples(args.source, args.allow_file, args.output, args.index)
    except (OSError, ValueError, csv.Error) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(f"kept={kept}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

--index list keeps the original list; --index set builds one set after reading the allow-file. Both modes share is_allowed(), the input loop, and output handling.

This exercise accepts exactly the sample_id,note header and nonempty IDs. Strings are preserved: each allow-file line is one ID, without additional whitespace trimming. The input uses the same record-by-record CSV approach suitable for larger files, while the allow-list remains in memory.

4. Profile the list version

1. Display help, then save statistics to list.prof:

python3 filter_samples.py --help
python3 -m cProfile -o list.prof filter_samples.py \
  samples.csv allowed.txt list.csv --index list

Expect kept=10000. list.csv contains the filtered records; list.prof contains profiling data rather than a text report. -o overwrites an existing profile. Read and analyze it in the same Python environment.

2. Sort by cumulative and internal time, then inspect callers of the membership function:

python3 - <<'PY_STATS'
from pstats import Stats, SortKey

stats = Stats("list.prof").strip_dirs()
print("=== cumulative time ===")
stats.sort_stats(SortKey.CUMULATIVE).print_stats(12)
print("=== function's own time ===")
stats.sort_stats(SortKey.TIME).print_stats(12)
print("=== membership call sites ===")
stats.print_callers("is_allowed")
PY_STATS

The Python profiler documentation defines ncalls as call count, tottime as time excluding subcalls, and cumtime as time including subcalls. Cumulative time helps locate entry points; internal time helps locate work. Do not add parent and child cumulative times together.

Expect 20000 calls to is_allowed(). List membership happens inside it; it was the main hotspot in this test. Seconds, source line numbers, and ranking vary by environment. If a report only shows that main() is slow, inspect its subfunctions or use print_callers() to locate a call site.

5. Build one set and compare the hotspot

Run the set version and compare the same function:

python3 -m cProfile -o set.prof filter_samples.py \
  samples.csv allowed.txt set.csv --index set
python3 - <<'PY_COMPARE'
from pstats import Stats, SortKey

for name in ("list.prof", "set.prof"):
    print(f"=== {name} ===")
    Stats(name).strip_dirs().sort_stats(SortKey.TIME).print_stats("is_allowed")
PY_COMPARE

Both runs should print kept=10000, with 20000 calls to is_allowed(). The work per lookup changes: a list compares elements in sequence, while a set provides a hash index for repeated membership checks. Do not call set(allowed_ids) inside is_allowed(), which would rebuild the index repeatedly.

The set type documentation describes sets as collections of distinct hashable elements. Here, strings are hashable and removing duplicate allowed IDs preserves existence checks. Sets do not preserve the original list order, but the CSV input drives iteration, so output order stays intact. A requirement to sort by the allow-list or count its duplicates would need different logic.

6. Check the output, then time the real command

1. Check bytes, record count, order, repeated IDs, and a multiline note:

python3 - <<'PY_CHECK'
import csv
from pathlib import Path

assert Path("list.csv").read_bytes() == Path("set.csv").read_bytes()
with open("set.csv", encoding="utf-8", newline="") as file:
    reader = csv.reader(file)
    assert next(reader) == ["sample_id", "note"]
    rows = list(reader)
assert len(rows) == 10000
assert [row[0] for row in rows] == [
    f"S{i % 10000:05d}" for i in range(20000) if i % 2 == 0
]
assert rows[0][1] == "记录 0, item\nsecond line"
assert sum(row[0] == "S00000" for row in rows) == 2
print("OK: identical bytes, 10000 rows, order and repeated IDs preserved")
PY_CHECK

Expect OK: identical bytes, 10000 rows, order and repeated IDs preserved. Loading rows into a list here is only for checking the small demonstration output.

2. Run each mode three times without cProfile and report the median:

python3 - <<'PY_TIME'
from pathlib import Path
from statistics import median
import subprocess
import sys
import tempfile
from time import perf_counter

with tempfile.TemporaryDirectory(prefix="cprofile-timing-") as directory:
    for index in ("list", "set"):
        elapsed = []
        for run in range(3):
            output = Path(directory) / f"{index}-{run}.csv"
            start = perf_counter()
            subprocess.run([
                sys.executable, "filter_samples.py", "samples.csv",
                "allowed.txt", str(output), "--index", index,
            ], check=True, stdout=subprocess.DEVNULL)
            elapsed.append(perf_counter() - start)
            assert output.read_bytes() == Path("list.csv").read_bytes()
        print(f"{index}: median_seconds={median(elapsed):.6f}")
PY_TIME

perf_counter() measures elapsed time. This check includes interpreter startup, file I/O, and filtering. Its temporary output directory is cleaned up automatically. Three local runs help check this change; fixed list-before-set ordering, file caches, and server load affect results, so they do not establish a universal speedup.

Profiling adds overhead. Use it to locate functions and compare whole-command timing without it. If memory rises after optimization, use tracemalloc snapshot comparisons to examine the index’s memory cost.

7. Troubleshooting

This example profiles the current process; it does not automatically inspect subprocess internals. It writes output from the beginning and does not guarantee a complete file after a mid-run error. Add an output commit strategy before adapting the controlled exercise to a production pipeline.

Summary

Capture function statistics, identify repeated work, then change the relevant data structure. Preserve input order and repeated records, compare complete outputs, and measure command time with profiling disabled. For this filter, constructing one set of allowed IDs removes repeated list scans.