Split a CSV File in Python and Keep the Header in Every Part

Download the scripts and tests (ZIP)
Python scripts, tests, and ready-to-run merge, join, and date practice files with expected results. No signup required.

An import tool accepts at most 10,000 records per upload, but your export contains more. Split the export into smaller CSV files with a header in each part. This guide uses a streaming script: it reads one parsed record at a time instead of loading the full table into a list.

Tested: Python 3.10 on Windows, September 7, 2026. The accompanying tests check record counts, repeated headers, leading-zero IDs, Unicode, quoted commas and embedded newlines, exact chunk boundaries, header-only input, invalid rows, and refusal to reuse an existing output directory. No extra packages or network requests are involved. This is a row-count splitter; it does not promise a maximum file size in bytes.

Why counting text lines breaks some exports

A quoted CSV field can contain a newline. That single record occupies more than one physical line. Cutting the text every 10,000 lines can separate the opening and closing quote and leave two invalid files. We use Python’s CSV parser to find record boundaries and its writer to serialize each complete record. See the official CSV reader and writer documentation for the file-opening convention and parser options.

Try five invented records first

Save this as orders.csv using UTF-8. Keep the line break inside the quoted note:

id,note
001,"Seoul, Korea"
002,"first line
second line"
003,café
004,서울
005,finished

There are five data records, even though the file has seven physical lines including its header. Splitting into two records per file should produce three parts containing 2, 2, and 1 records respectively. Each part also gets a header.

Complete script

Save the following as split_csv.py beside your sample, or use the copy in the ZIP’s examples folder.

"""Split UTF-8 CSV records, repeat headers, and mark successful completion."""
import argparse
import csv
import json
from pathlib import Path


def split_csv(source, destination, rows_per_file):
    if rows_per_file < 1:
        raise ValueError('rows_per_file must be positive')
    source, destination = Path(source), Path(destination)
    with source.open(encoding='utf-8-sig', newline='') as stream:
        reader = csv.reader(stream, strict=True)
        header = next(reader, None)
        if not header or any(not field.strip() for field in header):
            raise ValueError('A nonempty header is required')
        if len(set(header)) != len(header):
            raise ValueError('Duplicate header names')
        # Refuse an existing directory, even if it is empty.
        destination.mkdir()
        count, parts, output = 0, 0, None
        try:
            for row in reader:
                if len(row) != len(header):
                    raise ValueError(f'Wrong field count near physical line {reader.line_num}')
                if count % rows_per_file == 0:
                    if output is not None:
                        output.close()
                    parts += 1
                    output = (destination / f'part-{parts:06d}.csv').open(
                        'x', encoding='utf-8', newline='')
                    writer = csv.writer(output)
                    writer.writerow(header)
                writer.writerow(row)
                count += 1
        finally:
            if output is not None:
                output.close()
        result = {'complete': True, 'records': count, 'parts': parts,
                  'rows_per_file': rows_per_file}
        with (destination / 'manifest.json').open('x', encoding='utf-8') as manifest:
            json.dump(result, manifest, indent=2)
        return result


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('source', type=Path)
    parser.add_argument('destination', type=Path)
    parser.add_argument('--rows', type=int, required=True)
    args = parser.parse_args()
    try:
        result = split_csv(args.source, args.destination, args.rows)
    except (OSError, ValueError, csv.Error) as exc:
        parser.exit(1, f'Error: {exc}\nDo not use incomplete output. Retry in a new directory.\n')
    print(json.dumps(result))


if __name__ == '__main__':
    main()

Run from that folder:

python split_csv.py orders.csv output-parts --rows 2

The destination must not already exist, and its parent folder must exist. The script creates output-parts, three CSV files, and a completion manifest. It leaves the source file untouched. A successful command prints:

{"complete": true, "records": 5, "parts": 3, "rows_per_file": 2}

The first part, part-000001.csv, contains records 001 and 002. The embedded newline stays inside the quoted note. The second part contains 003 and 004; the third contains 005. CSV quoting and line endings can change during writing, so compare parsed field values rather than expecting byte-identical text.

For an actual import limit of 10,000 data records, use a fresh destination:

python split_csv.py orders.csv import-parts --rows 10000

If the importing system counts its header toward a 10,000-row limit, use --rows 9999 instead. Check the destination tool’s own requirements. A limit in megabytes needs a different splitting approach because field lengths vary.

Verify before importing

A zero exit status and a valid manifest.json with complete: true indicate that this script finished its pass. They do not establish that your source data is correct or that the receiving system will accept it. Review the record count against the export and test an import of one part first.

The following independent check compares the parsed records of the sample with the ordered output parts. Save it as verify_sample.py beside the sample and run python verify_sample.py:

import csv
from pathlib import Path

with open('orders.csv', encoding='utf-8-sig', newline='') as f:
    original = list(csv.reader(f))
combined = []
for path in sorted(Path('output-parts').glob('part-*.csv')):
    with path.open(encoding='utf-8', newline='') as f:
        rows = list(csv.reader(f))
    assert rows[0] == original[0]
    combined.extend(rows[1:])
assert combined == original[1:]
print('All five sample records preserved')

This small verification example loads the data into memory; use it for the sample, not as a memory-efficient validator for a huge export. The download contains automated tests that perform the same record-preservation check and exercise failure paths.

Failure cases and recovery

Situation Result and next step
Destination exists The run fails without reusing it. Choose a new directory name.
Empty file or blank/duplicate header The run fails before creating the destination. Correct the input schema.
Only a header Success with zero records and zero parts; only the manifest is created.
Record has too many or too few fields The run stops. Earlier parts may remain but there is no completed manifest.
Broken quoting, wrong encoding, disk error, interruption The command may leave partial output. Do not import it; fix the cause and retry in a new directory.
A single very long field exceeds the parser limit Python reports a CSV error. Assess the export rather than assuming that a lower chunk size will fix it.

The script intentionally leaves partial files for inspection. It does not roll back a failed job or resume it. An interrupted manifest write can itself leave invalid JSON, so the existence of the filename alone is insufficient. Do not run multiple processes against the same destination or edit the input while a run is in progress.

Recovery example verified: Python 3.10 on Windows, September 25, 2026. The failing input, separate corrected input, fresh-directory retry, existing-directory refusal, and checker below were executed.

Reproduce a partial run, then recover safely

A failed run can leave CSV files that look ready to import. Reproduce this in a scratch directory: save the following as bad-orders.csv in UTF-8. The last record deliberately has an extra field.

id,note
001,first
002,second
003,third,unexpected

Run:

python split_csv.py bad-orders.csv failed-parts --rows 2

Expect exit status 1 and these messages on standard error (standard output is empty):

Error: Wrong field count near physical line 4
Do not use incomplete output. Retry in a new directory.

failed-parts/part-000001.csv contains the header and records 001 and 002, but failed-parts/manifest.json does not exist. Importing the available part would silently omit the third record. Keep it for diagnosis, not import.

For this invented example, the extra field is known to be a mistake. Save a separate fixed-orders.csv rather than overwriting the failing input:

id,note
001,first
002,second
003,third

Run python split_csv.py fixed-orders.csv recovered-parts --rows 2. Expect exit status 0 and {"complete": true, "records": 3, "parts": 2, "rows_per_file": 2}. For real data, establish the intended schema before removing any field. Retrying into failed-parts will fail because the destination already exists, even if the source is now valid.

Save this as verify_recovery.py beside the files and run python verify_recovery.py without Python’s -O option. It checks the partial-output warning signs and the recovered records:

import csv
import json
from pathlib import Path

def read_rows(path):
    with Path(path).open(encoding="utf-8-sig", newline="") as stream:
        return list(csv.reader(stream))

assert not Path("failed-parts/manifest.json").exists()
assert read_rows("failed-parts/part-000001.csv") == [
    ["id", "note"], ["001", "first"], ["002", "second"]
]
manifest = json.loads(Path("recovered-parts/manifest.json").read_text(encoding="utf-8"))
assert manifest == {"complete": True, "records": 3, "parts": 2, "rows_per_file": 2}
parts = sorted(Path("recovered-parts").glob("part-*.csv"))
assert [p.name for p in parts] == ["part-000001.csv", "part-000002.csv"]
combined = []
for path, count in zip(parts, [2, 1]):
    rows = read_rows(path)
    assert rows[0] == ["id", "note"]
    assert len(rows) - 1 == count
    combined.extend(rows[1:])
assert combined == [["001", "first"], ["002", "second"], ["003", "third"]]
print("PASS: partial output identified; three recovered records verified")

Expected output: PASS: partial output identified; three recovered records verified. This checker is specific to the small example. It does not validate arbitrary production exports or make partial results safe to use.

Encoding, memory, and spreadsheet limits

Input is UTF-8, optionally with a BOM; output is UTF-8 without a BOM. Comma delimiters and standard double-quote escaping are assumed. Semicolon-delimited or legacy-encoded files need an explicit adaptation. Blank physical lines outside quoted fields are treated as invalid rows rather than silently dropped.

The script preserves text values such as 001. A spreadsheet application can still reinterpret them when opening the output, so import identifier columns as text when that matters. Cells beginning with formula characters are also preserved; this script is not a spreadsheet formula sanitizer.

The splitter retains the header and current record rather than the whole export. Memory therefore depends on record size, and output consumes disk space in addition to the original. We have not benchmarked multi-gigabyte exports or network drives; the reported verification covers the included synthetic cases.

Need the opposite operation? Use the CSV merge guide. To compare exports by a stable ID after an import or update, see CSV snapshot comparison.

Scroll to Top