Dividir um CSV grande em arquivos menores sem perder o cabeçalho

01Defina o que conta como uma linha

O script cria arquivos CSV numerados contendo no máximo a quantidade escolhida de registros de dados. Cada arquivo começa com o mesmo cabeçalho. O cabeçalho não conta para o limite. Com um limite de 3 e 7 registros de dados, os arquivos contêm 3, 3 e 1 registros de dados.

Um registro CSV não corresponde necessariamente a uma linha física de texto. Um campo entre aspas pode conter uma quebra de linha. O módulo csv lê os registros de acordo com as regras de aspas do CSV, portanto uma observação multilinha permanece vinculada ao seu registro em vez de se tornar uma linha separada.

02Crie a entrada sintética

Este conjunto de dados é sintético e foi criado especificamente para este artigo. Salve-o como sample.csv na sua pasta de trabalho. Copie exatamente a quebra de linha dentro da observação entre aspas; ela foi incluída de propósito para testar um caso que uma divisão comum por linhas trata incorretamente.

record_id,team,units,note
R001,Support,12,"starter, pack"
R002,Ops,7,repeat
R003,Support,9,"two
lines"
R004,Sales,5,normal
R005,Ops,11,normal
R006,Sales,6,normal
R007,Support,10,normal

Há 4 colunas e 7 registros de dados. A vírgula em starter, pack pertence a um único campo. A note correspondente a R003 ocupa 2 linhas físicas, mas continua sendo um único campo. Os valores de units somam 60: 12 + 7 + 9 + 5 + 11 + 6 + 10.

03Prepare a pasta de trabalho

1. Coloque sample.csv e um novo script chamado split_large_csv.py na mesma pasta de trabalho.
2. Cole no script o código Python completo da próxima seção.
3. Mantenha ROWS_PER_FILE em 3 neste exemplo. O valor deve ser um inteiro positivo.
4. Abra um terminal na pasta de trabalho e execute o comando abaixo.

python split_large_csv.py

Caminhos relativos são resolvidos a partir do diretório de trabalho do terminal, e não automaticamente a partir da localização do script. A pasta principal outputs pode já existir, mas outputs/sample_parts não deve existir quando esta execução começar.

04Divida os registros e verifique os arquivos gravados

O divisor lê os registros de forma incremental e abre cada parte em modo de criação exclusiva. Depois da gravação, ele lê novamente a fonte e as partes para comparar cabeçalhos, valores dos campos, ordem dos registros e contagens. As mensagens de sucesso aparecem somente depois que essa comparação termina.

import csv
from pathlib import Path

SOURCE = Path("sample.csv")
OUTPUT_DIR = Path("outputs") / "sample_parts"
ROWS_PER_FILE = 3


def part_path(number: int) -> Path:
    return OUTPUT_DIR / f"part_{number:04d}.csv"


def split_csv() -> tuple[int, int]:
    if type(ROWS_PER_FILE) is not int or ROWS_PER_FILE < 1:
        raise ValueError("ROWS_PER_FILE must be a positive integer.")

    with SOURCE.open("r", encoding="utf-8-sig", newline="") as source:
        reader = csv.reader(source, strict=True)
        header = next(reader, None)
        if not header or any(not name.strip() for name in header):
            raise ValueError("Missing header or blank column name.")
        if len(set(header)) != len(header):
            raise ValueError("Duplicate column names.")

        OUTPUT_DIR.parent.mkdir(parents=True, exist_ok=True)
        OUTPUT_DIR.mkdir()  # Refuse to reuse an existing output path.
        total = 0
        parts = 0
        target = None
        try:
            for record_no, row in enumerate(reader, start=1):
                if len(row) != len(header):
                    raise ValueError(
                        f"Data record {record_no}: wrong number of fields."
                    )
                if total % ROWS_PER_FILE == 0:
                    if target is not None:
                        target.close()
                    parts += 1
                    target = part_path(parts).open(
                        "x", encoding="utf-8", newline=""
                    )
                    writer = csv.writer(target)
                    writer.writerow(header)
                writer.writerow(row)
                total += 1
        finally:
            if target is not None:
                target.close()

    return total, parts


def verify_csv(expected_rows: int, part_count: int) -> None:
    with SOURCE.open("r", encoding="utf-8-sig", newline="") as source:
        original = csv.reader(source, strict=True)
        header = next(original, None)
        seen = 0
        for number in range(1, part_count + 1):
            with part_path(number).open(
                "r", encoding="utf-8", newline=""
            ) as part:
                rows = csv.reader(part, strict=True)
                if next(rows, None) != header:
                    raise ValueError(f"Header mismatch in part {number}.")
                count = 0
                for row in rows:
                    if row != next(original, None):
                        raise ValueError(f"Data mismatch in part {number}.")
                    count += 1
                required = min(
                    ROWS_PER_FILE,
                    expected_rows - (number - 1) * ROWS_PER_FILE,
                )
                if count != required:
                    raise ValueError(f"Row count mismatch in part {number}.")
                seen += count
        if seen != expected_rows or next(original, None) is not None:
            raise ValueError("Overall row count mismatch.")


if __name__ == "__main__":
    total, parts = split_csv()
    verify_csv(total, parts)
    print(f"Verified {total} data records in {parts} files.")
    print(f"Output folder: {OUTPUT_DIR.as_posix()}")

A decodificação da entrada aceita UTF-8 com ou sem uma marca de ordem de bytes no início. A saída usa UTF-8 sem essa marca. O argumento newline permite que o módulo csv trate os finais de registro e as quebras de linha incorporadas.

05Confira manualmente o resultado esperado

Os arquivos esperados estão listados abaixo. Os totais de units são verificações manuais independentes, e não cálculos realizados pelo divisor. Todos os arquivos devem repetir record_id,team,units,note como cabeçalho.

O texto esperado no console é mostrado abaixo. Ele foi derivado manualmente e não é um log capturado de uma execução.

Verified 7 data records in 3 files.
Output folder: outputs/sample_parts

Confira se 28 + 22 + 10 é igual ao total original de 60. Abra a primeira parte em um editor de texto para inspecionar a vírgula entre aspas e a note multilinha. Contar apenas as linhas visíveis resultará em uma contagem incorreta dos registros de dados.

06Confira antes de usar uma entrada maior

1. Confirme que todos os 3 arquivos tenham os mesmos 4 campos de cabeçalho e que o cabeçalho apareça apenas uma vez em cada arquivo.
2. Leia os arquivos em ordem numérica. De R001 a R007, cada registro deve aparecer uma única vez, sem registros ausentes ou repetidos.
3. Confirme que o script informe sucesso na verificação. A comparação verifica as strings dos campos, não apenas as contagens, portanto contagens iguais por si só não são suficientes.
4. Execute o script novamente sem alterar o destino. Ele deve parar com FileExistsError antes de gravar outra parte.

Teste os limites em pastas de trabalho separadas: 6 registros devem gerar 2 partes de 3, e não uma terceira parte vazia. Uma entrada contendo somente o cabeçalho deve deixar uma pasta de saída vazia. Uma entrada completamente vazia deve ser rejeitada. Essas expectativas foram avaliadas por inspeção, e não executadas aqui.

07Reconheça erros comuns

Um erro ocorrido mais tarde pode deixar arquivos parciais. O script fecha o arquivo atual, mas não reverte a pasta nem exclui nada automaticamente.

08Entenda os limites antes de aumentar a escala

Depois que o exemplo funcionar corretamente na sua máquina, altere SOURCE, escolha um novo OUTPUT_DIR dentro de outputs e aumente ROWS_PER_FILE. Um limite de registros não é um limite de tamanho em bytes: campos longos podem fazer com que grupos com a mesma quantidade de registros ocupem espaços em disco muito diferentes.

O script preserva as strings dos campos já interpretados, e não os bytes originais, o estilo de aspas ou o formato de final de registro. Ele não carrega o conjunto de dados inteiro de uma vez, mas campos excepcionalmente grandes ainda exigem memória e podem ultrapassar o limite de tamanho de campo do analisador CSV. A verificação acrescenta outra leitura da fonte e das saídas. Mantenha a fonte inalterada durante todo o processo; isto não é um snapshot bloqueado nem um backup transacional.

Registro de execução e verificação

2026-09-20 · exemplo conferido manualmente · alvo: Python 3.12 · biblioteca padrão: csv, pathlib · sem execução

• Foram contadas manualmente 4 colunas e 7 registros de dados, tratando a note multilinha entre aspas como um único campo.
• Os registros foram distribuídos manualmente em grupos de 3, 3 e 1.
• Foram calculados manualmente os totais dos grupos de 28, 22 e 10, resultando em 60 no total.
• Foram inspecionados os nomes dos arquivos de saída, a criação exclusiva, a repetição do cabeçalho e a lógica de comparação sequencial.
• O texto esperado do console foi derivado a partir do exemplo e do código.

Princípios de redação e verificação de todo o site

Fontes de referência

As explicações e os exemplos são de elaboração própria. Os comportamentos e conceitos relacionados podem ser consultados nas fontes oficiais abaixo.

• Documentação oficial do Python 3.12 · csv ↗ (New tab)
• Documentação oficial do Python 3.12 · pathlib ↗ (New tab)
• Documentação oficial do Python 3.12 · open ↗ (New tab)