Find Duplicate CSV IDs with Python and SQLite

A CSV can pass a row-format check and still contain the same sample ID several times. That matters before joining labels, merging batches, or counting unique samples. A Python set is convenient for a small input, but it keeps one entry for every distinct ID in memory.

This tutorial stores those IDs in a temporary SQLite database instead. It reads the CSV one record at a time and reports repeated IDs with their counts and first and last record positions. The earlier streaming CSV tutorial counts records by label; this one adds a separate uniqueness audit.

1. Decide what counts as a duplicate

The input must be UTF-8 CSV, optionally with a byte-order mark. Its header must contain sample_id; other columns may appear in any order. Column names must be nonempty and unique, and each record must have the same number of fields as the header.

IDs are compared as exact text. 001 and 1 differ, as do A and a. The script rejects empty IDs and leading or trailing whitespace instead of silently trimming them. It does not normalize Unicode. Other field values are left unchecked.

The report groups records by ID only. Two rows with the same ID are duplicates for this audit even when their labels disagree. A repeated ID may indicate a legitimate replicate, a merge mistake, or conflicting metadata; review it before removing rows.

2. Download and check the environment

Download find_duplicate_ids.py into a new working directory. Use Python 3.10 or newer with sqlite3 available, and SQLite 3.24.0 or newer for UPSERT support. The example was tested on macOS with Python 3.14.7 and SQLite 3.53.4. No pip packages or database server are needed.

python3 --version
python3 -c "import sqlite3; print(sqlite3.sqlite_version)"
python3 find_duplicate_ids.py --help

Python’s sqlite3 module opens an on-disk database directly. Some custom Python builds omit this module; use a Python distribution that includes it if the import fails.

3. Run the duplicate audit

The following bash or zsh commands create a small synthetic input and a report. Use a scratch directory: cat > and > replace existing files with the same names. Never redirect the report to the input path.

cat > samples.csv <<'CSV'
sample_id,label
001,1
002,0
001,0
003,1
002,0
002,1
CSV
python3 find_duplicate_ids.py samples.csv > duplicates.csv
cat duplicates.csv

The script writes this summary to standard error:

records=6 unique_ids=3 duplicate_ids=2

The report on standard output contains:

sample_id,occurrences,first_record,last_record
001,2,1,3
002,3,2,6

Here, duplicate_ids=2 means two distinct IDs occur more than once. There are three extra occurrences beyond the first occurrence of each ID: 6 - 3 = 3. These quantities answer different questions.

Record numbers start at 1 after the header. They refer to parsed CSV records, not physical lines: a quoted field may contain a newline. The report is sorted by the ID’s SQLite BINARY text order, not numeric order or order of first appearance.

A valid file with no duplicates produces only the report header. A header-only input is also valid. A successful audit returns exit status 0 even when it finds duplicates; status 1 means an input, database, or I/O error.

4. The complete script

"""Find exact duplicate sample_id values in a UTF-8 CSV. Python 3.10+."""
import argparse
from contextlib import closing
import csv
from pathlib import Path
import sqlite3
import sys
import tempfile

UPSERT = """
INSERT INTO ids VALUES (?, 1, ?, ?)
ON CONFLICT(sample_id) DO UPDATE SET
    occurrences = ids.occurrences + 1,
    last_record = excluded.last_record
"""


def audit(input_path, output, *, work_dir=None):
    """Validate all input before writing a CSV report; return audit counts."""
    with tempfile.TemporaryDirectory(prefix="csv-ids-", dir=work_dir) as directory:
        with closing(sqlite3.connect(Path(directory) / "ids.sqlite3")) as db:
            db.execute("""CREATE TABLE ids (
                sample_id TEXT PRIMARY KEY NOT NULL,
                occurrences INTEGER NOT NULL,
                first_record INTEGER NOT NULL,
                last_record INTEGER NOT NULL
            )""")
            records = 0
            with db, open(input_path, encoding="utf-8-sig", newline="") as handle:
                reader = csv.reader(handle, strict=True)
                header = next(reader, None)
                if not header or any(not name for name in header):
                    raise ValueError("Expected a nonempty CSV header")
                if len(set(header)) != len(header) or "sample_id" not in header:
                    raise ValueError("Header must be unique and include sample_id")
                column = header.index("sample_id")
                for records, row in enumerate(reader, start=1):
                    if len(row) != len(header):
                        raise ValueError(f"Record {records}: wrong field count")
                    sample_id = row[column]
                    if not sample_id or sample_id != sample_id.strip():
                        raise ValueError(f"Record {records}: empty or padded sample_id")
                    db.execute(UPSERT, (sample_id, records, records))
            unique_ids = db.execute("SELECT COUNT(*) FROM ids").fetchone()[0]
            writer = csv.writer(output, lineterminator="\n")
            writer.writerow(["sample_id", "occurrences", "first_record", "last_record"])
            duplicate_ids = 0
            for row in db.execute("""SELECT * FROM ids WHERE occurrences > 1
                                     ORDER BY sample_id COLLATE BINARY"""):
                writer.writerow(row)
                duplicate_ids += 1
    return records, unique_ids, duplicate_ids


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input", type=Path)
    parser.add_argument("--work-dir", type=Path, help="existing directory for scratch database")
    args = parser.parse_args()
    try:
        records, unique_ids, duplicates = audit(args.input, sys.stdout, work_dir=args.work_dir)
    except (OSError, UnicodeError, ValueError, csv.Error, sqlite3.Error) as exc:
        parser.exit(1, f"error: {exc}\n")
    print(f"records={records} unique_ids={unique_ids} duplicate_ids={duplicates}", file=sys.stderr)


if __name__ == "__main__":
    main()

The CSV reader handles quoting and embedded line breaks. Opening with newline="" lets it handle CSV newlines, while utf-8-sig accepts an optional initial BOM. See Python’s CSV documentation.

The table keeps one entry per distinct ID. SQLite’s UPSERT syntax increments the count when the primary key already exists and updates the last record number. The first record number stays unchanged. SQL placeholders pass IDs as values, including IDs containing quotes.

The input loop runs in one transaction. The database context manager commits on success or rolls back on an exception; closing(...) separately closes the connection before the temporary directory is removed. See the connection context-manager documentation.

The report starts only after all input has been parsed and validated. This prevents a malformed late record from producing an apparently complete report. A write failure during report output can still leave a partial file. Shell redirection also creates or truncates the report before Python starts, so this is not an atomic output-file writer.

5. Verify counts and error handling

Save this as check_report.py beside duplicates.csv, then run python3 check_report.py:

import csv

with open("duplicates.csv", encoding="utf-8", newline="") as handle:
    rows = list(csv.DictReader(handle))
assert rows == [
    {"sample_id": "001", "occurrences": "2", "first_record": "1", "last_record": "3"},
    {"sample_id": "002", "occurrences": "3", "first_record": "2", "last_record": "6"},
]
print("PASS: duplicate counts and record positions match")

Now create a malformed input and run the audit directly:

cat > invalid.csv <<'CSV'
sample_id,label
001,1
002
CSV
python3 find_duplicate_ids.py invalid.csv

Expected error, with exit status 1 and no CSV on standard output:

error: Record 2: wrong field count

The second data record has one field instead of two. Fix the source file and rerun; do not treat an empty redirected report as a clean result.

6. Choose where the temporary database lives

The default temporary directory may be a small partition or a RAM-backed filesystem. On Linux, /tmp is sometimes tmpfs, so a file there can still consume memory. Choose a work directory on a suitable disk-backed filesystem when moving the ID index out of RAM matters.

This example uses a scratch directory on the current filesystem:

mkdir csv-scratch
python3 find_duplicate_ids.py samples.csv --work-dir csv-scratch > duplicates-second.csv

The directory must already exist. Each run creates its own csv-ids-... subdirectory and normally removes it when finished, including after a handled input error. Forced termination can leave scratch files behind. The temporary database is not a saved audit artifact; keep the CSV report and an input checksum manifest for traceability.

This approach avoids a Python collection that grows with the number of unique IDs. It still uses memory for the current record, the header, SQLite’s page cache, and I/O buffers; it is not a hard memory cap. The database and its index require storage that grows with the distinct IDs. The 100,000-record test checks correctness, not a speed or peak-memory claim.

7. Troubleshooting

Use a stable input file: the script does not lock it or take a snapshot. For a table small enough to fit comfortably in memory, an in-memory method may be simpler. This tutorial makes no claim that SQLite is always faster.

8. Summary

Stream the CSV and keep the growing ID index in a temporary SQLite database on a suitable filesystem. Check the audit’s exit status, then review repeated IDs using the reported record positions. Decide how to resolve duplicates separately from detecting them.