A Repeatable Way to Clean a Messy CSV File With Python

Cleaning a CSV is one of those jobs that is easy to do once and hard to do twice. The first time you open the file, find the problems and fix them with a handful of commands, and the result is good. Next month a new file arrives with the same problems and two new ones, and nobody remembers the commands, or the order they ran in, or what the numbers were supposed to add up to. A repeatable cleaning process is a small program that takes the raw file and produces a clean file, a list of what it refused to keep, and a report of what it did. It is the same every time you run it, and it never edits the original. This guide builds one for a messy customer export, step by step, and ties together the eleven articles in this series that go deep on each step. Everything below ran, and the numbers are the ones I got.

The data is synthetic: a hand-written 9 row file with each defect once, and a generated file of 200,000 rows with the defects at realistic rates.

What is wrong with the file

Before the process, the inventory. These are the defects in the 200,000 row export, with the article that treats each one:

Eleven kinds of problem in one file is not unusual. Exports that pass through a spreadsheet, a web form and two systems pick them all up.

Seven rules that make it repeatable

The point is not any single fix. It is the discipline around them, and these are the rules I follow.

1. Never edit the raw file. The input is read-only. The cleaner writes new files and records the input's checksum (sha256 prefix a25cf1988f29218c in the report). When someone asks what the data looked like before, you have it. When a rule turns out to be wrong, you fix the rule and run again from the original.

2. Read everything as text. pd.read_csv(path, dtype=str, keep_default_na=False). No type inference, no NA conversion, no leading zeros lost. Every conversion afterwards is a line in your code, which you can read and test, and not a decision a library made for you (what libraries decide).

3. One transformation per step, in a fixed order. Each step does one thing, so that when a count is wrong you know which step to look at. The order matters, as the next section shows.

4. Count rows in and out of every step. If a step drops rows, say how many and why. If it changes fields, say how many. These counts are the report, and they are the first thing to compare between runs.

5. Keep what you reject. Dropped rows go to a file with a reason, not into the void. Parse failures become empty values with the original text kept, and unparseable values are counted.

6. Do not guess. If a value is ambiguous (03/04/2025), unknown (N/A) or unparseable, the cleaned value is empty and the problem is reported. Cleaning that invents values is data fabrication with good intentions.

7. Make the run deterministic and testable. Same input, same output, byte for byte. Unit tests for every step function, so you can change the rules without fear.

The order of the steps

The order is not arbitrary. Each step depends on the one before it:

1. Read as text, remembering each row's line number in the file.

2. Drop non-data rows (blank, repeated header, footer). They would fail every later step, and the footer is a free checksum.

3. Trim whitespace and invisible characters. Everything that compares values (dedupe, joins, mappings) needs this first.

4. Convert missing tokens to real missing values, keeping the reason.

5. Parse dates, money, integers, phones and names, each with its own function and its own failure count.

6. Normalize categorical values through mapping tables, and report the unmapped.

7. Deduplicate on the normalized key, with a stated keeper rule.

8. Check the result against patterns and invariants, and fail if they do not hold.

Swapping steps 3 and 7 is the classic mistake: dedupe on untrimmed, mixed-case emails and 3,032 of the 6,070 duplicates survive. Swapping 2 and 5 sends a Total row through the date parser and counts it as an error. The pipeline below follows this order exactly.

The pipeline

The functions that do each step are small and sit in clean.py (documented in the pandas and Python re references) (parse_money, parse_date, parse_int, norm_phone, norm_country and a few more). The pipeline calls them in order, counts, and writes the outputs:

    df = read_raw(src)
    missing = [c for c in REQUIRED if c not in df.columns]
    if missing:
        raise CleaningError(f"{report['source']}: missing columns {missing}; found {list(df.columns)}")
    df["line"] = df.index + 2
    step("read as text", len(df), len(df))

    body = df.drop(columns="line")
    blank = body.apply(lambda c: c.str.strip()).eq("").all(axis=1)
    repeat = body["id"].str.strip().str.lower().eq("id")
    footer = body["id"].str.strip().str.match(r"(?i)^(total\b|\(\d+ rows?\))")
    is_data = ~(blank | repeat | footer)
    df = df[is_data].copy()
    step("drop blank, repeated header and footer rows", len(is_data), len(df), f"{int(blank.sum())} blank, {int(repeat.sum())} header, {int(footer.sum())} footer")

The first block reads the file and drops the non-data rows. The mask for a blank row is "every field is empty after stripping", the repeated-header mask compares the first cell with id, and the footer is a pattern, all of which are explained in the blank rows article. Then the trimming, with the invisible-character regex:

    text_cols = [c for c in df.columns if c != "line"]
    padded = int(sum((df[c] != df[c].str.strip()).sum() for c in text_cols))
    for c in text_cols:
        df[c] = df[c].str.replace(INVISIBLE, "", regex=True).str.replace(r"\s+", " ", regex=True).str.strip()
    step("trim whitespace and invisible characters", len(df), len(df), f"{padded} fields had edge whitespace")

    out = pd.DataFrame({"id": df["id"], "line": df["line"]})
    names = df["name"]
    lf = names.str.contains(",", regex=False)
    comma = names.str.split(",", n=1, expand=True)
    space = names.str.split(" ", n=1, expand=True)
    space[1] = space[1].fillna("")
    out["first_name"] = comma[1].str.strip().where(lf, space[0]).str.title()
    out["last_name"] = comma[0].str.strip().where(lf, space[1]).str.title()
    out["email"] = df["email"].str.lower()
    phone = df["phone"].map(norm_phone)
    out["phone"] = phone.fillna("")
    out["phone_raw_unparsed"] = (df["phone"] != "") & phone.isna()
    out["phone_unparsed"] = df["phone"].where(out["phone_raw_unparsed"], "")
    dates = df["signup_date"].map(parse_date)
    out["signup_date"] = dates.map(lambda d: d.isoformat() if d else "")
    money = df["balance"].map(parse_money)
    out["balance"] = money.map(lambda v: f"{v:.2f}" if v is not None else "")
    out["balance_currency"] = df["balance"].str.extract(r"([$€])")[0].map(CURRENCY).fillna("")
    out["units"] = [("" if (n := parse_int(v)) is None else str(n)) for v in df["units"]]
    out["country"] = df["country"].map(norm_country).fillna("")
    out["status"] = df["status"].str.casefold()

Names are split by form (Last, First versus First Last), emails lowercased, phones converted to E.164, dates to ISO, money to a two-decimal string with a separate currency column taken from the symbol, integers parsed with the thousands rule, countries mapped to ISO codes and statuses casefolded. The last block deduplicates and writes:

    bad_dates = int((out["signup_date"] == "").sum())
    if max_bad_dates is not None and bad_dates / len(out) > max_bad_dates:
        raise CleaningError(f"{bad_dates} of {len(out)} signup dates did not parse ({bad_dates / len(out):.2%}), above the allowed {max_bad_dates:.2%}")
    out["_signup"] = pd.to_datetime(out["signup_date"].replace("", pd.NA))
    ordered = out.sort_values(["_signup", "line"], na_position="first")
    kept = ordered.drop_duplicates("email", keep=keep)
    dropped = ordered[ordered.duplicated("email", keep=keep)].copy()
    winner = kept.set_index("email")["id"]
    dropped["reason"] = "duplicate email; kept id " + dropped["email"].map(winner)
    final = kept.sort_values("line").drop(columns=["_signup", "line", "phone_raw_unparsed"])
    step(f"deduplicate on email, keep {'latest' if keep == 'last' else 'earliest'} signup", len(out), len(final), f"{len(dropped)} rows dropped")

    final.to_csv(f"{out_dir}/{report['source'].replace('.csv', '')}.clean.csv", index=False)
    dropped.drop(columns=["_signup", "phone_raw_unparsed"]).to_csv(f"{out_dir}/{report['source'].replace('.csv', '')}.dropped.csv", index=False)
    report["rows_out"] = len(final)
    report["seconds"] = round(time.perf_counter() - t0, 2)
    json.dump(report, open(f"{out_dir}/{report['source'].replace('.csv', '')}.report.json", "w"), indent=1)

It sorts by parsed signup date (and then file line), keeps the latest row for each email, and writes the losers to dropped.csv with a reason that names the winner. The keeper rule is a decision, and it is written in one line where a reader can see it and change it (choosing the keeper).

What it did

On the small file, which has one of each defect:

python /batches/csv-0001-0200/lab/clean/clean_customers.py customers-messy.csv /tmp/cleaned && cat /tmp/cleaned/customers-messy.clean.csv && echo --- dropped && cat /tmp/cleaned/customers-messy.dropped.csv

customers-messy.csv (sha256 88b1f9b6fbda2853): 8 clean rows in 0.03s
  read as text                                                    11 ->      11   0.01s  
  drop blank, repeated header and footer rows                     11 ->       9   0.01s  1 blank, 0 header, 1 footer
  trim whitespace and invisible characters                         9 ->       9   0.00s  3 fields had edge whitespace
  parse names, phones, dates, money, integers, countries           9 ->       9   0.01s  unparsed: 1 dates, 1 balances, 2 phones
  deduplicate on email, keep latest signup                         9 ->       8   0.00s  1 rows dropped
  fields changed by parsing: {'name': 2, 'email': 2, 'phone': 8, 'signup_date': 4, 'balance': 9, 'units': 6, 'country': 9, 'status': 3}
id,first_name,last_name,email,phone,phone_unparsed,signup_date,balance,balance_currency,units,country,status
2,Jane,Smith,[email protected],+14155550188,,2025-03-15,0.00,USD,85,US,active
3,Bo,Chen,[email protected],+14155550101,,2025-03-15,-12.50,,,US,active
4,Cy,Dubois,[email protected],+14155550177,,2025-03-16,15.75,EUR,1050,US,inactive
5,Ana,Silva,[email protected],+14155550132,,2025-03-20,1300.00,USD,1200,US,active
6,Di,Rossi,[email protected],,555-0100 ext. 12,2025-03-17,1234.50,EUR,900,IT,active
7,Ed,Zürich,[email protected],,,,7.00,USD,,CH,pending
8,Flo,Kiev,[email protected],,+380 44 123 4567,2025-03-18,,,,UA,pending
9,Gus,Riga,[email protected],+14155550199,,2025-03-19,20.00,USD,5,US,active
--- dropped
id,line,first_name,last_name,email,phone,phone_unparsed,signup_date,balance,balance_currency,units,country,status,reason
1,2,Ana,Silva,[email protected],+14155550132,,2025-03-14,1234.50,USD,1200,US,active,duplicate email; kept id 5

The log shows the five steps with rows in and out. Eleven lines became nine after removing a blank row and the Total footer; three fields had edge whitespace; one date, one balance and two phone numbers could not be parsed; one row was dropped as a duplicate. The cleaned table is easy to check by eye. Ana Silva appears once, as id 5, the later of two signups (id 1 went to dropped.csv with duplicate email; kept id 5). SMITH, JANE is Jane Smith. Di Rossi's balance of 1.234,50 € is 1234.50 in EUR. Bo Chen's (12.50) is -12.50 with no currency, because the source gave none. Ed Zürich's unparseable date is empty. And the two phone numbers the US-only parser could not handle, the extension number and the Ukrainian one, are kept as text in phone_unparsed and not thrown away.

Now the same pipeline on 200,000 rows:

python /batches/csv-0001-0200/lab/clean/clean_customers.py customers-messy-200k.csv /tmp/cleaned

customers-messy-200k.csv (sha256 a25cf1988f29218c): 193708 clean rows in 3.85s
  read as text                                                200001 ->  200001   0.31s  
  drop blank, repeated header and footer rows                 200001 ->  199778   0.10s  222 blank, 0 header, 1 footer
  trim whitespace and invisible characters                    199778 ->  199778   0.39s  32950 fields had edge whitespace
  parse names, phones, dates, money, integers, countries      199778 ->  199778   2.42s  unparsed: 959 dates, 11972 balances, 0 phones
  deduplicate on email, keep latest signup                    199778 ->  193708   0.25s  6070 rows dropped
  fields changed by parsing: {'name': 19986, 'email': 3065, 'phone': 166475, 'signup_date': 113167, 'balance': 170265, 'units': 159191, 'country': 199778, 'status': 99723}

200,001 lines in, 193,708 rows out, in 3.85 s. The steps tell the story: 222 blank rows and 1 footer dropped (the file has 200,001 lines including the header, 199,778 data rows); 32,950 padded fields trimmed; 959 dates and 11,972 balances unparsed (the balances are the missing tokens, not failures); 0 phones unparsed; 6,070 duplicates removed. Every one of those numbers matches the per-step articles, which ran the same functions separately.

The report is the product

The pipeline also writes report.json, a record of the input checksum, each step's counts and the elapsed time:

head -c 700 /tmp/cleaned/customers-messy-200k.report.json

{
 "source": "customers-messy-200k.csv",
 "sha256": "a25cf1988f29218c",
 "steps": [
  {
   "step": "read as text",
   "rows_in": 200001,
   "rows_out": 200001,
   "seconds": 0.31,
   "note": ""
  },
  {
   "step": "drop blank, repeated header and footer rows",
   "rows_in": 200001,
   "rows_out": 199778,
   "seconds": 0.09,
   "note": "222 blank, 0 header, 1 footer"
  },
  {
   "step": "trim whitespace and invisible characters",
   "rows_in": 199778,
   "rows_out": 199778,
   "seconds": 0.39,
   "note": "32950 fields had edge whitespace"
  },
  {
   "step": "parse names, phones, dates, money, integers, countries",
   "rows_in": 199778,
   "rows_out": 199778,
   "seconds": 2.45,
   "note": "u

Keep it with the output. When next month's run says 32950 padded fields and this one says 640,000, something upstream changed, and you know before you look at the data. Compare reports between runs: a sudden change in the count dropped at any step is a better alarm than a failed job.

Checking the output independently

A cleaner that grades its own work proves little, so I check the output with a different tool, DuckDB, reading the clean file as text:

cd /tmp/cleaned
duckdb -box -c "
SELECT count(*) AS rows,
       count(DISTINCT email) AS distinct_emails,
       count(*) FILTER (WHERE first_name <> trim(first_name) OR last_name <> trim(last_name)) AS padded_names,
       count(*) FILTER (WHERE signup_date IS NOT NULL AND NOT regexp_full_match(signup_date, '\d{4}-\d{2}-\d{2}')) AS bad_dates,
       count(*) FILTER (WHERE phone IS NOT NULL AND NOT regexp_full_match(phone, '\+1\d{10}')) AS bad_phones,
       count(*) FILTER (WHERE country IS NULL) AS no_country,
       count(*) FILTER (WHERE balance IS NULL) AS no_balance,
       count(*) FILTER (WHERE signup_date IS NULL) AS no_date,
       count(*) FILTER (WHERE phone IS NULL) AS no_phone
FROM read_csv('customers-messy-200k.clean.csv', all_varchar=true)"
duckdb -box -c "SELECT balance_currency, count(*) AS rows, round(sum(try_cast(balance AS DECIMAL(18,2))), 2) AS total FROM read_csv('customers-messy-200k.clean.csv', all_varchar=true) GROUP BY 1 ORDER BY 1"
duckdb -box -c "SELECT status, count(*) AS rows FROM read_csv('customers-messy-200k.clean.csv', all_varchar=true) GROUP BY 1 ORDER BY 2 DESC"

┌────────┬─────────────────┬──────────────┬───────────┬────────────┬────────────┬────────────┬─────────┬──────────┐
│  rows  │ distinct_emails │ padded_names │ bad_dates │ bad_phones │ no_country │ no_balance │ no_date │ no_phone │
├────────┼─────────────────┼──────────────┼───────────┼────────────┼────────────┼────────────┼─────────┼──────────┤
│ 193708 │ 193708          │ 0            │ 0         │ 0          │ 0          │ 11617      │ 902     │ 32273    │
└────────┴─────────────────┴──────────────┴───────────┴────────────┴────────────┴────────────┴─────────┴──────────┘
┌──────────────────┬───────┬──────────────┐
│ balance_currency │ rows  │    total     │
├──────────────────┼───────┼──────────────┤
│ EUR              │ 52093 │ 507672444.27 │
│ USD              │ 78260 │ 763946673.12 │
│ NULL             │ 63355 │ -1815243.08  │
└──────────────────┴───────┴──────────────┘
┌──────────┬───────┐
│  status  │ rows  │
├──────────┼───────┤
│ active   │ 97198 │
│ inactive │ 64154 │
│ pending  │ 32356 │
└──────────┴───────┘

The first table: 193,708 rows and 193,708 distinct emails (so no duplicates remain), 0 padded names, 0 dates that do not match the ISO pattern, 0 phones that do not match +1 and ten digits. It also reports what is still empty, on purpose: 11,617 rows without a balance, 902 without a signup date and 32,273 without a phone. Those are unknowns, not errors, and the cleaned file says so instead of inventing values.

The second table is the one to read slowly. The balances sum to 507,672,444.27 in euros (52,093 rows), 763,946,673.12 in dollars (78,260 rows) and -1,815,243.08 for the 63,355 rows with no currency symbol, which includes the parenthesized negatives and the plain numbers. Adding those three numbers together would produce a figure with no meaning, and the cleaned file's job is to make that visible instead of hiding it in one column. Whether the unlabeled rows are dollars, euros or something else is a question for the data owner, and the answer belongs in the contract.

Same input, same output

Repeatability is a property you can test. I ran the pipeline twice from the raw file and hashed the clean file each time, and hashed the raw file:

python /batches/csv-0001-0200/lab/clean/clean_customers.py customers-messy-200k.csv /tmp/cleaned > /dev/null
sha256sum /tmp/cleaned/customers-messy-200k.clean.csv | cut -c1-16
python /batches/csv-0001-0200/lab/clean/clean_customers.py customers-messy-200k.csv /tmp/cleaned2 2>/dev/null || (mkdir -p /tmp/cleaned2 && python /batches/csv-0001-0200/lab/clean/clean_customers.py customers-messy-200k.csv /tmp/cleaned2 > /dev/null)
sha256sum /tmp/cleaned2/customers-messy-200k.clean.csv | cut -c1-16
sha256sum customers-messy-200k.csv | cut -c1-16

8411365bdfdabc3b
customers-messy-200k.csv (sha256 a25cf1988f29218c): 193708 clean rows in 3.96s
  read as text                                                200001 ->  200001   0.32s  
  drop blank, repeated header and footer rows                 200001 ->  199778   0.10s  222 blank, 0 header, 1 footer
  trim whitespace and invisible characters                    199778 ->  199778   0.39s  32950 fields had edge whitespace
  parse names, phones, dates, money, integers, countries      199778 ->  199778   2.43s  unparsed: 959 dates, 11972 balances, 0 phones
  deduplicate on email, keep latest signup                    199778 ->  193708   0.25s  6070 rows dropped
  fields changed by parsing: {'name': 19986, 'email': 3065, 'phone': 166475, 'signup_date': 113167, 'balance': 170265, 'units': 159191, 'country': 199778, 'status': 99723}
8411365bdfdabc3b
a25cf1988f29218c

Both clean outputs have the SHA-256 prefix 8411365bdfdabc3b, and the raw file is still a25cf1988f29218c, the same value the report recorded. Determinism needs a few habits: no timestamps in the output, a stable sort with a tie-break (the pipeline orders by signup date and then by file line), and no dependence on dictionary or set order. Sorting by a column with ties and keeping "the last" gives a different winner on a different run unless the tie is broken.

Changing the keeper rule and seeing the cost

The deduplication step takes the keeper as an option, keep=last by default and keep=first on request, because "which duplicate wins" is the decision most likely to be argued about later. Running the same 200,000 rows with keep=first and comparing the two outputs in DuckDB shows what the argument is about:

python /batches/csv-0001-0200/lab/clean/clean_customers.py customers-messy-200k.csv /tmp/cleaned_first keep=first 2>/dev/null || (mkdir -p /tmp/cleaned_first && python /batches/csv-0001-0200/lab/clean/clean_customers.py customers-messy-200k.csv /tmp/cleaned_first keep=first | head -n 1)
cd /tmp
duckdb -box -c "
WITH a AS (SELECT * FROM read_csv('cleaned/customers-messy-200k.clean.csv', all_varchar=true)),
     b AS (SELECT * FROM read_csv('cleaned_first/customers-messy-200k.clean.csv', all_varchar=true))
SELECT (SELECT count(*) FROM a) AS rows_latest, (SELECT count(*) FROM b) AS rows_earliest,
       (SELECT count(*) FROM a ANTI JOIN b USING (id)) AS ids_only_in_latest,
       (SELECT count(*) FROM b ANTI JOIN a USING (id)) AS ids_only_in_earliest,
       (SELECT round(sum(try_cast(balance AS DECIMAL(18,2))), 2) FROM a) AS balance_latest,
       (SELECT round(sum(try_cast(balance AS DECIMAL(18,2))), 2) FROM b) AS balance_earliest"

customers-messy-200k.csv (sha256 a25cf1988f29218c): 193708 clean rows in 3.79s
  read as text                                                200001 ->  200001   0.28s  
  drop blank, repeated header and footer rows                 200001 ->  199778   0.10s  222 blank, 0 header, 1 footer
  trim whitespace and invisible characters                    199778 ->  199778   0.39s  32950 fields had edge whitespace
  parse names, phones, dates, money, integers, countries      199778 ->  199778   2.43s  unparsed: 959 dates, 11972 balances, 0 phones
  deduplicate on email, keep earliest signup                  199778 ->  193708   0.23s  6070 rows dropped
  fields changed by parsing: {'name': 19986, 'email': 3065, 'phone': 166475, 'signup_date': 113167, 'balance': 170265, 'units': 159191, 'country': 199778, 'status': 99723}
┌─────────────┬───────────────┬────────────────────┬──────────────────────┬────────────────┬──────────────────┐
│ rows_latest │ rows_earliest │ ids_only_in_latest │ ids_only_in_earliest │ balance_latest │ balance_earliest │
├─────────────┼───────────────┼────────────────────┼──────────────────────┼────────────────┼──────────────────┤
│ 193708      │ 193708        │ 5907               │ 5907                 │ 1269803874.31  │ 1270240268.56    │
└─────────────┴───────────────┴────────────────────┴──────────────────────┴────────────────┴──────────────────┘

Both outputs have 193,708 rows, because the number of distinct emails does not depend on which row represents each one. They do not contain the same rows. 5,907 ids appear only in the latest-signup output and a different 5,907 only in the earliest-signup output, and the balance columns differ by 436,394.25 in total (1,269,803,874.31 against 1,270,240,268.56). That total adds euro, dollar and unlabeled rows together, so it is not a meaningful figure on its own; it is there to show that the choice moves money-valued columns, not just ids. Put that comparison in front of whoever owns the data before you pick, and write the answer in the report, not only in the code.

Failing early on a schema change

A pipeline that carries on after its input changed shape can produce a clean-looking file that is wrong. The cheapest protection is a check on the header before any work is done. I renamed signup_date to signup in the small file and ran it again:

mkdir -p /tmp/renamed && sed '1s/signup_date/signup/' customers-messy.csv > /tmp/renamed/customers-messy.csv
python /batches/csv-0001-0200/lab/clean/clean_customers.py /tmp/renamed/customers-messy.csv /tmp/cleaned; echo "exit status: $?"

cleaning stopped: customers-messy.csv: missing columns ['signup_date']; found ['id', 'name', 'email', 'phone', 'signup', 'balance', 'units', 'country', 'status']
exit status: 2

The run stopped before writing anything, said which column was missing and which ones it found, and exited with status 2. The exit status is the part a scheduler reads. Use a distinct code for "the input is not what I expect" so a job runner can tell it from a crash, and have the message name what was found, because the usual cause is a changed export setting rather than a bug in your code.

Failing on a threshold

Some defects are expected and some are alarms. 959 dates out of 199,778 that do not parse is 0.48%, which I consider normal for this generated source and which is reported in the run. The same count at 20% would mean the source changed its date format. A threshold turns that judgment into a rule that runs every time:

python /batches/csv-0001-0200/lab/clean/clean_customers.py customers-messy-200k.csv /tmp/cleaned max_bad_dates=0.01 | head -n 1; echo "exit status: ${PIPESTATUS[0]}"
python /batches/csv-0001-0200/lab/clean/clean_customers.py customers-messy-200k.csv /tmp/cleaned max_bad_dates=0.001; echo "exit status: $?"

customers-messy-200k.csv (sha256 a25cf1988f29218c): 193708 clean rows in 3.8s
exit status: 0
cleaning stopped: 959 of 199778 signup dates did not parse (0.48%), above the allowed 0.10%
exit status: 2

With max_bad_dates=0.01 (1%) the run finished normally and exited 0. With max_bad_dates=0.001 (0.1%) it stopped with 959 of 199778 signup dates did not parse (0.48%), above the allowed 0.10% and exit status 2, before the deduplication or any output was written. Two practical notes. Pick the threshold from several past runs of the report, not from one, and set it a little above what is normal so it catches change without crying wolf. And make the failure message carry both the count and the share, because "959" is meaningless without the denominator. The same idea applies to any count in the report: padded fields, duplicates, rows dropped. The validation guide turns that into a proper contract.

Reading the dropped file

The clean file is what you ship, and the dropped file is what you review. After the 200,000 row run, dropped.csv has 6,070 rows plus the header, and a grouped count in DuckDB shows one reason for all of them:

cd /tmp/cleaned
head -n 4 customers-messy-200k.dropped.csv | cut -d, -f1-4,12-
wc -l customers-messy-200k.dropped.csv
duckdb -box -c "SELECT split_part(reason, ';', 1) AS reason, count(*) AS rows FROM read_csv('customers-messy-200k.dropped.csv', all_varchar=true) GROUP BY 1"

id,line,first_name,last_name,country,status,reason
89,90,Cy,Chen,DE,active,duplicate email; kept id 34090
457,458,Jun,Chen,DE,pending,duplicate email; kept id 1658
850,851,Di,Zurich,IT,active,duplicate email; kept id 152932
6071 customers-messy-200k.dropped.csv
┌─────────────────┬──────┐
│     reason      │ rows │
├─────────────────┼──────┤
│ duplicate email │ 6070 │
└─────────────────┴──────┘

Each row keeps its original file line (line 90, 458, 851 above) and names the id that replaced it, so a reviewer can open the raw file at that line and compare the two records. For example, id 89 on line 90 lost to id 34090. Today there is a single reason because the pipeline has a single rule that discards good-looking rows. When you add a second rule, give it its own reason string, so the grouped count stays readable; a reject file with one generic reason is a reject file nobody can triage. I also sample a few dropped rows at random on every run and read both sides, because the most expensive mistake in this pipeline is a keeper rule that is wrong for a subset of the data, and that mistake is invisible in the clean file.

Tests for the cleaner

The step functions are the part that changes, and the part that breaks quietly. Sixteen unit tests cover them:

import sys
from datetime import date
from decimal import Decimal

import pandas as pd
import pytest

sys.path.insert(0, "/batches/csv-0001-0200/lab/clean")
from clean import parse_date, parse_int, parse_money, norm_phone, norm_country


@pytest.mark.parametrize("text, want", [
    ("$1,234.50", Decimal("1234.50")), ("(12.50)", Decimal("-12.50")), ("1.234,50 €", Decimal("1234.50")),
    ("$7,00", Decimal("7.00")), ("-", None), ("abc", None),
])
def test_money(text, want):
    assert parse_money(text) == want


@pytest.mark.parametrize("text, want", [
    ("2025-03-14", date(2025, 3, 14)), ("03/15/2025", date(2025, 3, 15)), ("15 Mar 2025", date(2025, 3, 15)),
    ("2025-3-6", date(2025, 3, 6)), ("31/02/2025", None), ("2025-13-45", None), ("TBD", None),
])
def test_dates(text, want):
    assert parse_date(text) == want


def test_thousands_and_missing():
    assert parse_int("1,200") == 1200 and parse_int("N/A") is None


def test_phone_and_country():
    assert norm_phone("(415) 555-0132") == norm_phone("415.555.0132") == "+14155550132"
    assert norm_country(" U.S. ") == "US" and norm_country("Atlantis") is None


def test_zero_width_space_is_removed_before_matching():
    s = pd.Series(["[email protected]​", "[email protected]"])
    cleaned = s.str.replace(r"[​‌‍⁠]", "", regex=True).str.lower()
    assert cleaned.nunique() == 1

cd /batches/csv-0001-0200/lab/clean && python -m pytest -q test_clean.py 2>&1 | tail -n 3

................                                                         [100%]
16 passed in 0.33s

16 passed in 0.33s. The cases are the ones that were bugs while I wrote this series: the European 1.234,50 €, the accounting negative, the date 31/02/2025 that matches a date shape and is not a date, 2025-13-45, the two phone spellings that must give the same number, and the zero-width space that must not make two emails different. Every time a customer's file teaches you a new failure, add a test with that value before you fix the code. The test suite is the memory the process needs.

What I would add before trusting it in production

The script here is the core, and a few more habits turn it into a job. Store each run's report beside its output under the input's checksum, so that two files with the same name but different contents never overwrite each other's results. Keep the mapping tables for countries, statuses and missing tokens in files that a non-programmer can read in a code review, since those are the rules that change most often. Write the output to a temporary name and rename it only when every check has passed, so a half-finished run can never be mistaken for a good one by whatever reads the folder. Record the version of the pipeline (a commit hash is enough) in the report, because the same input run through next year's rules will not produce the same file, and you should be able to say which rules made which output. None of these changes the cleaning logic. They make the surrounding process as repeatable as the transformation itself.

When it gets bigger

Pandas processed 200,000 rows in under 4 seconds, with the heaviest steps being the row-level Python functions (about 0.45 s for money and 0.9 s for dates). For tens of millions of rows, the same steps move into SQL: DuckDB reads the file as text, regexp_replace and strptime do the parsing, and a window function keeps the latest row per email, as in the duplicates article. The rules do not change, only where they run. If the file is too large to hold, stream it: the shell tools in working with CSV in the shell and the csv module in the Python guide process a row at a time.

Where the time goes

The report records seconds per step, which is how I know where to look if the job gets slow. For the 200,000 rows:

Parsing is 61% of the run, and within it the row-by-row Python functions (dates, money) dominate: the date article measured 0.91 s for 199,778 dates and the currency article 0.45 s for the balances. That is the right place to optimize if you ever need to: parse the dominant format with a vectorized to_datetime(format=...) and fall back to the function only for the rows that fail. Reading and trimming are cheap. At this size none of it matters, and I would not optimize before the job is a problem. What matters is that the cost is visible, per step, in every run.

How much the cleaning changed

The report also counts the fields the parsing step changed in each column, compared with the trimmed raw text:

fields changed by parsing: {'name': 19986, 'email': 3065, 'phone': 166475, 'signup_date': 113167,
  'balance': 170265, 'units': 159191, 'country': 199778, 'status': 99723}

Read it as a map of how dirty each column was. country changed in all 199,778 rows, because every value was mapped to an ISO code (even Brazil, to BR), so that count measures the mapping, not the mess. phone changed 166,475 times, the number of non-empty phones, since E.164 is a different representation from every input. name changed only 19,986 times, once for each Last, First name, and email 3,065 times (case and invisible characters). signup_date changed 113,167 times, the dates that were not already ISO plus the 959 that became empty. balance changed 170,265 times, 85% of rows, because of the symbols, commas, parentheses and the 11,972 missing tokens. status changed 99,723 times, the values with a capital letter. A column where this number is surprising is the first one to examine: if email had changed 100,000 times, something upstream would have started upper-casing addresses.

Running it on next month's file

The first run proves the rules. Operating the process is a small routine:

1. Run the pipeline on the new delivery. It writes the clean file, the dropped file and a report named for the input.

2. Compare the report with last month's. The row counts will differ, but the proportions should not: padded fields, dropped duplicates, unparsed dates and the share of each balance style. A proportion that moves by an order of magnitude is a change in the source.

3. Read the unmapped and unparsed values by distinct value, not by row. New spellings of a country, a new placeholder like TBD and a new date format are the usual finds. Add them to the mapping table or the format list, with a test, and rerun.

4. Run the independent checks and the tests. Fail the job if the checks fail.

5. Publish the clean file, the dropped file and the report together, and store the raw input untouched.

The rules live in code and mapping tables in a repository, so a change is a commit with a reason: "added TBD to missing tokens, 236 rows". Six months later you can answer why a value was cleaned as it was, which no sequence of spreadsheet edits can do.

Which tool for which step

Pandas is a good home for this pipeline at moderate sizes, and the steps are not tied to it. A rough map from what I measured in the series:

• Reading, trimming, dropping rows: any tool. The shell tools work if the file has no quoted fields, as the shell guide explains.

• Dates and numbers: a Python function per format is the clearest, and DuckDB's strptime and try_cast run it faster on big files (DuckDB dates).

• Phone numbers: a library. The regex accepted 43.8% more numbers than phonenumbers called valid.

• Duplicates: a window function. QUALIFY row_number() OVER (PARTITION BY ...) states the keeper rule where a reader sees it.

• Pattern checks: SQL or pandas, with line numbers kept.

Whatever the tools, the structure is the same: raw in, clean out, rejects out, report out.

How to tell it worked

Count: lines in the file, minus non-data rows, minus duplicates, equals rows out, and each number is in the report. Hash the raw file before and after and confirm it did not change. Run twice and compare the output hashes. Run DuckDB checks on the output (distinct keys, patterns, ranges). Pick ten rows at random from the clean file, find each in the raw file by its line number and read both. Look at the dropped file, not only the clean one. And run the tests.

What a cleaning script cannot decide

It cannot decide that 03/04/2025 is March, that a balance with no currency symbol is dollars, that two customers with the same email are one person, or that an empty phone is acceptable. It can apply a rule you chose consistently and show you the cost of the rule. The rules belong to the people who use the data, and the way to get their agreement is to show them the report: here is what was dropped, here is what is empty, here is what the three currency totals are. When the cleaned file is the input to something that cannot afford surprises, the next step is to validate it before you trust it, which turns the checks above into a contract the pipeline enforces on every run.