PythonでCSVを指定件数に分割：すべてのファイルにヘッダーを付ける

行数ではなくレコード数で分ける

取り込み先が「1ファイルにつきデータ2件まで」の場合を、小さな例で再現します。ヘッダーはこの件数に含めません。ただし実際のサービスがヘッダーを数えるかどうかは、相手の仕様を確認してください。

セル内に改行があるCSVをテキストの行数で切ると、値の途中でファイルが分かれます。先にCSVとして読み、1レコードを1件として扱います。

ヘッダーを付けて分割するコード

入力はUTF-8 CSVを読み込んだ文字列です。空ファイル・重複した列名・列数の違いは停止し、空行は読み飛ばします。出力先へ保存する前に全件を検査する小規模向けの実装です。

import csv
import io

def split_csv(text, size):
    if type(size) is not int or size < 1:
        raise ValueError("分割件数は1以上の整数にしてください")
    reader = csv.reader(io.StringIO(text, newline=""), strict=True)
    header = next(reader, None)
    if not header or any(not h.strip() for h in header):
        raise ValueError("有効なヘッダーが必要です")
    if len(set(header)) != len(header):
        raise ValueError("列名が重複しています")
    rows = [row for row in reader if row]
    if any(len(row) != len(header) for row in rows):
        raise ValueError("列数が一致しません")
    chunks = []
    for start in range(0, len(rows), size):
        output = io.StringIO(newline="")
        writer = csv.writer(output, lineterminator="\n")
        writer.writerow(header)
        writer.writerows(rows[start:start + size])
        chunks.append(output.getvalue())
    return chunks

text = 'id,note\n001,"上段\n下段"\n002,青\n003,赤\n'
chunks = split_csv(text, 2)
for number, chunk in enumerate(chunks, start=1):
    records = list(csv.reader(io.StringIO(chunk, newline="")))
    print(f"part_{number:03d}.csv: データ{len(records) - 1}件")
    assert records[0] == ["id", "note"]
assert list(csv.reader(io.StringIO(chunks[0])))[1][1] == "上段\n下段"
assert split_csv("id,note\n", 2) == []

期待する結果と保存方法

part_001.csv: データ2件
part_002.csv: データ1件

1件目の備考は2行に見えますが、分割の基準では1件です。端数は最後のファイルへ入り、空の末尾ファイルは作りません。ヘッダーだけの入力は空のリストを返します。

実データではsizeを必要な件数に変えます。返った各文字列を、番号付きの新規ファイルへ保存します。ファイルを"x"モードで作れば、同名ファイルの上書きを防げます。再実行時は新しい出力フォルダーを用意してください。

保存処理の途中で権限エラーや容量不足が起きると、一部のファイルだけが残る可能性があります。取り込み先へ渡す前に、すべての出力が揃っているか確認します。

分割後の確認

各ファイルをCSVとして読み、ヘッダーを除いた件数の合計が元データの件数と一致することを確認します。分割ファイルを番号順に戻したときに、元のレコード列と一致するかを見ると、欠落だけでなく順序も確かめられます。

再結合する場合は、各ファイルのヘッダーをデータとして混ぜないようにします。ヘッダーを1回だけ残す結合例が対応する処理です。

制限と変更が必要な場合

この関数は入力と出力をメモリーに保持します。巨大ファイルを処理するなら、readerから少しずつ読み、作業用ディレクトリーへ書き出す設計に変えてください。itertools.isliceは一定件数ずつ読む処理の候補ですが、途中失敗時の扱いも必要です。

件数ではなく「1ファイル何MB以下」という制限には、このコードだけでは対応できません。文字数とバイト数は違い、引用符や文字コードによってサイズも変わります。