큰 CSV를 일정한 행 수로 나누고 원본 ID와 내용을 검산하기
한 번에 올릴 수 있는 데이터가 4행으로 제한되어 있다면, CSV 10행을 4행·4행·2행의 세 파일로 나누고 다시 읽어 원본과 대조하세요. 파일 개수와 총행 수가 같아도 한 행을 빼고 다른 행을 두 번 넣었을 수 있습니다. 여기서는 각 파일의 헤더가 한 번씩 있는지, ID가 중복되지 않는지, 원본의 순서와 모든 필드가 그대로인지까지 검사합니다.
입력은 가상 문구류 10건입니다. 고객별로 묶거나 금액을 합치지 않고, 원래 순서대로 정한 행 수만큼 끊습니다. 큰 파일 전체를 목록으로 읽어 들이지 않는 스트리밍 방식이며 원본 파일은 쓰기 모드로 열지 않습니다.
1 입력과 완료 조건을 먼저 고정하기
실습 ZIP을 새 빈 연습 폴더에 직접 풀고, README.txt가 있는 폴더에서 작업하세요. 표준 라이브러리만 사용하며, 이 예제는 CPython 3.12.14와 Linux에서 검수했습니다. 다른 버전에서는 먼저 제공된 전체 테스트, 특히 메모의 CR·CRLF 보존 시험을 모두 통과해야 사용합니다. 다른 버전의 호환성을 검증한 원고는 아닙니다. 추가 패키지, 인터넷 연결, API 키는 필요하지 않습니다.
input/records.csv의 전체 내용은 다음과 같습니다. 직접 다시 저장하기보다 제공된 파일로 먼저 실행하세요.
id,item,memo
1,노트,일반
2,펜,"파랑, 검정"
3,파일,"첫 줄
둘째 줄"
4,봉투,"""확인"" 표시"
5,클립,납품 A
6,테이프,
7,상자,납품 B
8,라벨,선반 02
9,자,길이 확인
10,지우개,완료
ID 2의 메모에는 쉼표가 있고 ID 3의 메모에는 실제 줄바꿈이 있습니다. ID 3은 화면에서 두 줄로 보이지만 데이터 한 건입니다. ID 4의 따옴표도 메모 내용입니다. splitlines()로 물리 줄을 자르거나 쉼표마다 문자열을 나누면 이 값이 손상될 수 있습니다. csv.reader가 읽은 논리 레코드를 기준으로 나눕니다. Python 3.12 CSV 읽기와 줄바꿈 설명
이 코드의 입력 규칙은 다음과 같습니다.
- 헤더는 소문자
id,item,memo의 세 열이며 순서도 같아야 합니다 - 인코딩은 BOM 없는 UTF-8, 구분자는 쉼표, 인용 문자는 큰따옴표입니다. UTF-8 오류를 무시하거나 다른 인코딩을 추측하지 않습니다
- 각 데이터 레코드는 정확히 세 필드입니다. 빈 레코드, 열 부족·추가, 데이터 중간의 반복 헤더는 거부합니다
id는1,2,10처럼 앞자리 0 없는 양의 십진수 문자열입니다.01, 빈 ID, 중복 ID는 거부합니다item은 빈 문자열이면 거부합니다.memo는 빈 문자열을 허용합니다. 앞뒤 공백이나 메모의 줄바꿈을 지우지 않습니다- 필드 하나의 파서 한도는 1,000,000자로 정했습니다. 바이트 한도나 출력 파일의 용량 한도가 아닙니다
이 ID 규칙이 실제 자료와 다르면 원본 ID를 바꾸어 통과시키지 마세요. 실제 스키마에 맞게 검증 규칙과 테스트부터 별도로 바꾸어야 합니다. strict=True는 파서가 인식한 잘못된 인용 구조를 거부하며, 모든 CSV 방언을 검사하는 독립 문법 검증기는 아닙니다. Python CSV strict와 필드 크기 한도
예제의 완료 조건은 데이터 10건, 파일 3개, 각 데이터 수 [4, 4, 2], 헤더 수 [1, 1, 1], ID 순서 1~10, 중복 0입니다. 메모까지 포함한 세 필드의 값이 원본과 같아야 합니다. 헤더는 --max-rows의 행 수에 포함하지 않습니다. 행 수 제한을 맞춘 것이므로 파일의 바이트 용량 제한을 맞췄다고 판단하지 않습니다.
2 새 출력 폴더로 분할하기
터미널에서 실습 폴더를 작업 위치로 선택한 뒤 다음 명령을 실행합니다. 아래 python이 검수한 CPython 3.12.14를 가리키는지 먼저 확인하세요. 분할 전에 전체 자동시험을 실행하고, 하나라도 실패하면 분할 명령을 진행하지 마세요. 시스템에서 명령 이름이 python3이면 같은 명령의 첫 단어만 python3으로 바꿉니다.
python --version
python -m unittest discover -s tests -v
python split_csv.py split --source input/records.csv --output chunks4 --max-rows 4
chunks4는 아직 없어야 합니다. 코드는 출력의 상위 폴더를 새로 만들지 않으며, 기존 폴더·파일·정상 또는 끊어진 심볼릭 링크를 출력 이름으로 지정하면 중단합니다. 입력 파일과 출력 이름이 같거나 입력 파일 아래를 출력 경로로 지정해도 진행하지 않습니다. 입력과 출력 상위 경로의 링크·reparse 표식도 거부합니다.
통과한 결과 폴더에는 정확히 다음 네 파일이 생깁니다.
chunks4/
chunk_000001.csv 데이터 4건, ID 1·2·3·4
chunk_000002.csv 데이터 4건, ID 5·6·7·8
chunk_000003.csv 데이터 2건, ID 9·10
summary.json 재읽기 검산 결과
각 CSV의 첫 레코드는 id,item,memo입니다. chunk_000004.csv라는 빈 파일은 생기지 않습니다. 콘솔 출력과 chunks4/summary.json은 다음 내용입니다.
{
"rows": 10,
"chunks": 3,
"max_rows": 4,
"rows_per_chunk": [
4,
4,
2
],
"headers_per_chunk": [
1,
1,
1
],
"duplicates": 0,
"id_sequence_equal": true,
"field_values_equal": true,
"source_sha256": "961a45aa41dbead9203a56b1e144c5b17623fc81b742fdae351c2fb85c6d01e3"
}
source_sha256은 제공 원본의 바이트 지문입니다. 다른 업무 파일에는 다른 값이 나옵니다. 원본 바이트 보존과 결과의 필드 보존은 구분해서 보세요. 출력은 UTF-8과 LF 레코드 줄바꿈으로 다시 쓰므로 인용 방식이나 레코드 줄바꿈 바이트까지 원본과 같아야 하는 분할 도구는 아닙니다. 따옴표 안에 들어 있던 메모의 쉼표·줄바꿈·따옴표 값은 그대로 대조합니다.
3 결과를 순서대로 연결해 원본과 비교하기
분할 후에는 같은 입력과 같은 행 수 상한으로 별도 검산 명령을 실행하세요.
python split_csv.py verify --source input/records.csv --output chunks4 --max-rows 4
검산은 파일 이름의 번호 순으로 각 헤더를 확인한 뒤 헤더를 제외한 데이터만 이어서 읽습니다. 이어 읽은 결과와 원본을 첫 번째 레코드부터 마지막까지 일대일 비교합니다. zip_longest를 사용하므로 한쪽이 먼저 끝나도 누락이나 추가를 놓치지 않습니다. 파일 번호가 빠졌거나 중간 파일이 덜 찼거나 마지막에 빈 파일이 붙었을 때도 중단합니다. Python 3.12 zip_longest
정상이라면 분할 때와 동일한 JSON이 나옵니다. 결과에서 레코드 순서를 바꾸거나 메모 하나만 수정해도 검산에 실패합니다. summary.json이 없거나 저장된 결과와 현재 검산 결과가 다르면 완성본으로 인정하지 않습니다. 숫자만 적힌 보고서를 읽는 것이 아니라 CSV 파일을 실제로 다시 읽는 검사입니다.
분할 코드 자체도 임시 폴더에 쓴 CSV를 이 방식으로 재검산합니다. 통과한 뒤 summary.json을 쓰고 마지막에만 새 출력 폴더 이름으로 바꿉니다.
4 ZIP과 동일한 전체 코드
다음이 split_csv.py의 전체 내용입니다. 제공 ZIP에도 동일한 파일이 들어 있습니다.
"""Stream fixed-schema UTF-8 CSV chunks and verify ordered field conservation."""
import argparse
import csv
import hashlib
from itertools import zip_longest
import json
import os
from pathlib import Path
import re
import stat
import sys
from tempfile import TemporaryDirectory
HEADER = ['id', 'item', 'memo']
FIELD_LIMIT = 1_000_000
csv.field_size_limit(FIELD_LIMIT)
def plain_path(value, directory=False):
path = Path(os.path.abspath(value))
for part in [*reversed(path.parents), path]:
info = part.lstat()
if stat.S_ISLNK(info.st_mode) or getattr(info, 'st_file_attributes', 0) & 0x400:
raise ValueError(f'Link/reparse path is not allowed: {part}')
info = path.stat()
valid = stat.S_ISDIR(info.st_mode) if directory else stat.S_ISREG(info.st_mode)
if not valid:
raise ValueError(f'Expected a plain {"directory" if directory else "file"}: {path}')
return path
def sha256_file(path):
digest = hashlib.sha256()
with path.open('rb') as handle:
for block in iter(lambda: handle.read(1024 * 1024), b''):
digest.update(block)
return digest.hexdigest()
def records(path):
plain_path(path)
with path.open(encoding='utf-8', errors='strict', newline='') as handle:
reader = csv.reader(handle, strict=True)
if next(reader, None) != HEADER:
raise ValueError(f'{path.name}: expected BOM-free header id,item,memo')
for number, row in enumerate(reader, 1):
if len(row) != 3:
raise ValueError(f'{path.name}: record {number}: expected 3 fields')
if not re.fullmatch(r'[1-9][0-9]*', row[0]):
raise ValueError(f'{path.name}: record {number}: id must be a positive decimal without leading zeros')
if not row[1]:
raise ValueError(f'{path.name}: record {number}: item is empty')
yield row
def verify(source, folder, max_rows, expected_hash=None, require_summary=True):
if max_rows < 1:
raise ValueError('max-rows must be a positive integer')
source = plain_path(source)
folder = plain_path(folder, directory=True)
parts = []
for path in folder.iterdir():
if path.name == 'summary.json':
plain_path(path)
continue
match = re.fullmatch(r'chunk_([0-9]{6,})\.csv', path.name)
if not match:
raise ValueError(f'Unexpected output entry: {path.name}')
plain_path(path)
parts.append((int(match[1]), path))
parts.sort()
for index, (number, path) in enumerate(parts, 1):
if number != index or path.name != f'chunk_{index:06d}.csv':
raise ValueError('Chunk numbering is not contiguous from 1')
counts = []
def merged_records():
for index, (_, path) in enumerate(parts, 1):
count = 0
for row in records(path):
count += 1
if count > max_rows:
raise ValueError(f'{path.name}: too many data records')
yield row
if count == 0 or (index < len(parts) and count != max_rows):
raise ValueError(f'{path.name}: empty or short non-final chunk')
counts.append(count)
seen = set()
total = 0
missing = object()
for original, output in zip_longest(records(source), merged_records(), fillvalue=missing):
total += 1
if original is missing or output is missing or original != output:
raise ValueError(f'Ordered field comparison failed at record {total}')
if original[0] in seen:
raise ValueError(f'Duplicate id: {original[0]}')
seen.add(original[0])
digest = sha256_file(source)
if expected_hash is not None and digest != expected_hash:
raise ValueError('Source bytes changed during the run')
report = {
'rows': total, 'chunks': len(parts), 'max_rows': max_rows,
'rows_per_chunk': counts, 'headers_per_chunk': [1] * len(parts),
'duplicates': 0, 'id_sequence_equal': True, 'field_values_equal': True,
'source_sha256': digest,
}
summary = folder / 'summary.json'
if require_summary and not summary.exists():
raise ValueError('Completed output requires summary.json')
if summary.exists():
saved = json.loads(summary.read_text(encoding='utf-8'))
if saved != report:
raise ValueError('Saved summary does not match this source and output')
return report
def split(source, output, max_rows):
if max_rows < 1:
raise ValueError('max-rows must be a positive integer')
source = plain_path(source)
output = Path(os.path.abspath(output))
parent = plain_path(output.parent, directory=True)
if os.path.lexists(output):
raise ValueError(f'Output already exists; choose a new name: {output.name}')
before_hash = sha256_file(source)
with TemporaryDirectory(prefix=f'.{output.name}.incomplete-', dir=parent) as temp:
stage = Path(temp)
handle = None
seen = set()
total = 0
part = 0
try:
for row in records(source):
if row[0] in seen:
raise ValueError(f'Duplicate id: {row[0]}')
seen.add(row[0])
if total % max_rows == 0:
if handle is not None:
handle.close()
part += 1
handle = (stage / f'chunk_{part:06d}.csv').open(
'x', encoding='utf-8', newline='')
writer = csv.writer(handle, lineterminator='\n')
writer.writerow(HEADER)
writer.writerow(row)
total += 1
finally:
if handle is not None:
handle.close()
# Release the split-pass ID set before the verification pass.
del seen
report = verify(source, stage, max_rows, expected_hash=before_hash, require_summary=False)
with (stage / 'summary.json').open('x', encoding='utf-8', newline='') as handle:
json.dump(report, handle, ensure_ascii=False, indent=2)
handle.write('\n')
# Single-process check only: this is not a concurrency lock.
if os.path.lexists(output):
raise ValueError('Output appeared during work; choose a new name')
stage.rename(output)
return report
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('action', choices=['split', 'verify'])
parser.add_argument('--source', type=Path, required=True)
parser.add_argument('--output', type=Path, required=True)
parser.add_argument('--max-rows', type=int, required=True)
args = parser.parse_args()
try:
action = split if args.action == 'split' else verify
report = action(args.source, args.output, args.max_rows)
except (ValueError, csv.Error, UnicodeError, OSError) as error:
print(f'ERROR: {error}', file=sys.stderr)
return 2
print(json.dumps(report, ensure_ascii=False, indent=2))
return 0
if __name__ == '__main__':
raise SystemExit(main())
records()는 CSV를 한 레코드씩 돌려줍니다. split()은 다음 데이터가 실제로 도착했을 때만 새 파일을 열고 헤더를 한 번 씁니다. 그래서 데이터 수가 상한의 정확한 배수여도 꼬리의 빈 파일이 만들어지지 않습니다.
전체 데이터 행을 메모리에 올리지는 않지만, ID 중복 검사에 쓰는 seen 집합은 레코드 수 N에 비례하는 O(N) 메모리가 필요합니다. 결과 파일 목록과 행 수 목록도 파일 개수만큼 메모리를 씁니다. 수천만 ID에서 일정한 메모리 사용량을 보장하는 예제가 아닙니다. 이 규모라면 중복 검사를 로컬 데이터베이스나 디스크 기반 정렬로 옮긴 별도 설계가 필요합니다. 또 분할 후 재읽기와 원본 해시 검사 때문에 파일을 여러 번 읽습니다. 대용량 처리 속도는 측정하지 않았습니다.
5 경계 조건과 실패 복구를 연습하기
같은 원본에 상한만 바꾸면 다음 결과가 나와야 합니다. 각각 새 출력 이름을 사용하세요.
python split_csv.py split --source input/records.csv --output chunks1 --max-rows 1
python split_csv.py split --source input/records.csv --output chunks10 --max-rows 10
python split_csv.py split --source input/eight_records.csv --output chunks8 --max-rows 4
python split_csv.py split --source input/header_only.csv --output chunks0 --max-rows 4
chunks1: 데이터 1건짜리 CSV 10개chunks10: 데이터 10건짜리 CSV 1개, 끝의 빈 파일 없음chunks8: 데이터 4건짜리 CSV 2개, 끝의 빈 파일 없음chunks0: CSV 0개와summary.json만 생성. 올바른 헤더만 있는 입력은 이 예제에서 허용합니다
완전히 빈 파일은 올바른 헤더도 없으므로 거부합니다. 헤더만 있는 파일과 빈 파일은 다릅니다.
후반부 인용문 오류를 직접 확인하려면 다음 명령을 실행합니다. 이 입력은 앞의 정상 8건 뒤에 닫히지 않은 인용문이 붙어 있습니다.
python split_csv.py split --source input/failures/late_unclosed_quote.csv --output bad_chunks --max-rows 4
ERROR: unexpected end of data와 종료 코드 2가 나와야 합니다. 앞의 두 CSV를 임시로 썼더라도 처리된 파싱 오류에서는 임시 폴더를 정리하고 bad_chunks라는 완성 경로를 만들지 않습니다. 잘못된 입력을 고칠 때는 원본 근거를 확인하고 새 수정본을 만드세요. 오류 행을 몰래 제외하고 성공으로 처리하지 않습니다.
- 기존 출력 오류: 결과를 지우지 말고
--output chunks4_again처럼 새 이름을 지정합니다. 같은chunks4로 재실행하면 거부하는 것이 정상입니다 - 헤더·열 수 오류: 입력의 열 이름과 순서를 확인합니다. 필드를 자동으로 채우거나 버리지 않습니다
- 중복 ID: 원본 기록에서 동일 ID의 의미를 확인합니다. 이 예제는 파일을 넘어서 발견한 중복도 중단합니다
- 인코딩·인용문 오류: 새 사본에서 저장 형식과 따옴표 닫힘을 확인합니다. 오류를 무시하는 옵션은 없습니다
- 링크·상위 폴더 오류: 링크나 업무 공유 폴더 대신 새 로컬 연습 폴더에 제공 자료를 직접 풀고 실행합니다
- 검산 실패: 결과를 전달하지 말고 원본과 새 결과를 다시 생성·대조합니다. 변경된 결과를 기존
summary.json에 맞추려고 고치지 않습니다
TemporaryDirectory는 정상 종료나 처리 가능한 예외에서 임시 폴더 정리를 돕습니다. 파일 권한이 갑자기 바뀌어 정리에 실패하면 남은 임시 자료도 성공 결과로 사용하지 마세요. Python 3.12 TemporaryDirectory
다른 프로그램이 파일을 바꾸지 않는 개인 로컬 연습 사본에서 한 번에 한 작업만 실행하세요. 확정 직전에 기존 출력 경로를 다시 검사하지만 검사와 이름 변경 사이의 경쟁 생성까지 막는 잠금은 없습니다. POSIX의 이름 변경은 이미 있는 대상을 대체할 수 있으므로 동시 실행·동기화·공유 폴더에 이 예제를 그대로 쓰지 않습니다. 강제 종료나 전원 장애에서 임시 폴더 정리·저장장치 복구를 보장하지 않습니다. 강제 종료 뒤 남은 .이름.incomplete-... 폴더를 완성 이름으로 바꾸지 말고, 보존한 채 새 연습 폴더에서 다시 시작하세요. Python 3.12 Path.rename
CSV의 필드를 임의로 치환하지 않으므로 외부 메모가 수식처럼 보이는 문자열이어도 그대로 남습니다. 결과는 먼저 텍스트 편집기로 확인하고, 출처를 모르는 CSV를 Excel에서 더블클릭해 직접 열지 마세요. 이 실습에서 Windows나 Microsoft Excel 앱을 실행한 결과는 없습니다.
6 제공 자료로 같은 검사를 다시 실행하기
python -m unittest discover -s tests -v
python tests/run_fixture_checks.py
제공 자동시험은 48개입니다. 4·4·2 분할, 최대 1·10건, 정확한 배수, 헤더만 있는 입력, 헤더·열 수·인용문·UTF-8 오류, 후반 파싱 실패, 파일을 넘는 중복 ID, 기존 폴더·파일·링크, 재실행, 원본 바이트 보존과 결과 변조 검출, 메모 안의 단독 CR 및 CRLF 값 보존을 확인했습니다. 제공 입력 파일을 직접 돌리는 별도 검사는 19개 시나리오이며, 모든 입력의 실행 전후 SHA-256이 같았습니다. 테스트는 격리된 임시 폴더를 사용합니다.
expected/chunks4/는 비교 기준이고 actual/chunks4/는 실제 실행에서 보관한 결과입니다. 새로 만든 chunks4와 대조하세요. logs/에는 테스트 전 실패 기록, 구현 후 통과 기록, fixture별 실제 실행 결과와 Python 버전이 들어 있습니다. 이 기록은 가상 실습의 검산 범위를 보여 주며, 임의 업무 파일의 성공이나 운영 환경의 안전을 보증하지 않습니다.
실습 자료
P16_Practice_CSV_Chunks.zip에는 전체 split_csv.py, 동일한 입력 10건, 8건·헤더 전용 경계 입력, 잘못된 입력 14개, expected/의 고정 기대값, actual/chunks4/의 실제 세 CSV와 검산 보고서, README.txt, 48개 자동시험과 fixture 검사, 실행 로그·버전·SHA-256 목록을 담았습니다. ZIP은 직접 풀고 README부터 확인하세요. 새로 생성할 chunks4 폴더는 ZIP의 최상위에 넣지 않았습니다.
여러 Excel 파일을 먼저 모아야 하는 작업은 pandas로 Excel 파일 합치고 검산하기에서 이어 볼 수 있습니다. CSV의 열 수 오류 위치를 먼저 찾아야 한다면 CSV 논리 레코드와 물리 줄의 열 수 검사를 참고하세요.
공식 자료와 확인일
- Python 3.12 CSV
- Python 3.12 zip_longest
- Python 3.12 TemporaryDirectory
- Python 3.12 pathlib lstat와 rename
공식 자료 확인일은 2026-10-05입니다. 실제 실행 버전은 Python 3.12.14이며, 확인한 3.12 계열 공식 문서 페이지의 현재 표제는 3.12.15입니다. 다른 Python 버전과 Windows에서의 실행 시험은 하지 않았습니다. 다른 버전은 사용 전에 제공된 전체 테스트와 CR·CRLF 보존 시험을 먼저 통과해야 합니다.
작성자 소개
