파이썬 파일입출력 NDJSON 스트리밍 처리와 에러 행 분리 저장 방법

📂 파이썬 파일입출력 기본 이해

파이썬에서 파일 입출력은 데이터를 다루는 가장 기본적인 기술이면서도, 실제 개발 과정에서는 매우 자주 사용되는 기능입니다.
간단히 말해, 파일을 읽고 쓰는 과정을 통해 프로그램이 외부 데이터를 활용하거나 결과를 저장할 수 있게 되는 것이죠.
텍스트 파일, CSV, JSON 등 다양한 포맷이 있으며, 각각의 포맷에 맞게 데이터를 처리하는 방법을 알아두는 것이 중요합니다.

파일 입출력의 기본 구조는 크게 세 단계로 나눌 수 있습니다.
첫 번째는 파일 열기, 두 번째는 데이터 읽기 또는 쓰기, 그리고 마지막은 파일 닫기입니다.
이 흐름만 기억하면 대부분의 파일 작업은 어렵지 않게 다룰 수 있습니다.

• 📥open() 함수를 이용해 파일 열기
• ✍️read() 또는 write() 메서드로 데이터 처리
• ✅모든 작업이 끝나면 close() 메서드로 파일 닫기

하지만 매번 파일을 닫는 것을 잊어버리면 데이터 손상이나 메모리 누수와 같은 문제가 생길 수 있습니다.
이를 방지하기 위해 with 구문을 사용하는 것이 좋은 습관입니다.
with 구문은 블록이 끝나면 자동으로 파일을 닫아주기 때문에 안정적인 파일 처리가 가능합니다.

# 기본 파일 읽기 예제
with open("example.txt", "r", encoding="utf-8") as f:
    data = f.read()
    print(data)

# 기본 파일 쓰기 예제
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("파일에 새로운 내용을 작성합니다.")

이처럼 파일 입출력은 단순히 데이터를 읽고 쓰는 것을 넘어서, 더 복잡한 데이터 파이프라인을 구축하는 기초가 됩니다.
NDJSON 스트리밍이나 에러 행 분리 저장 같은 고급 기법 역시 이런 기초 위에서 이해할 수 있습니다.

📑 NDJSON 포맷의 특징과 활용 사례

NDJSON(Newline Delimited JSON)은 줄바꿈 문자로 각 JSON 객체를 구분하는 형식입니다.
기존 JSON은 하나의 큰 배열로 데이터를 담는 경우가 많아 전체를 한꺼번에 읽고 메모리에 로드해야 하지만, NDJSON은 레코드 단위로 처리할 수 있기 때문에 대용량 데이터 처리에 적합합니다.

NDJSON은 로그 데이터, 스트리밍 데이터, 이벤트 처리와 같은 상황에서 널리 활용됩니다.
예를 들어, 대규모 로그 파일에서 특정 조건에 맞는 레코드만 추출하거나, 실시간으로 수집되는 센서 데이터를 저장할 때 효과적입니다.
또한, 빅데이터 플랫폼과 연동할 때도 NDJSON은 많이 사용되는 포맷 중 하나입니다.

💬 NDJSON은 한 줄마다 독립적인 JSON 객체로 구성되기 때문에, 스트리밍 환경에서 데이터를 조금씩 읽고 처리하기에 이상적입니다.

다음은 NDJSON 파일의 구조를 간단히 예로 보여드립니다.
보시는 것처럼 JSON 배열처럼 대괄호나 쉼표 없이, 각 줄마다 독립된 JSON 객체가 기록됩니다.

{"id": 1, "message": "첫 번째 로그"}
{"id": 2, "message": "두 번째 로그"}
{"id": 3, "message": "세 번째 로그"}

NDJSON의 장점은 다음과 같습니다.

• ⚡한 줄씩 처리 가능하여 메모리 효율성이 높음
• 📊실시간 로그 분석이나 데이터 스트리밍에 최적화
• 🔗빅데이터 도구(Spark, Elasticsearch 등)와 호환성이 뛰어남

즉, NDJSON은 단순히 데이터 저장 포맷이 아니라, 효율적인 데이터 스트리밍과 분석을 위한 강력한 도구라 할 수 있습니다.
다음 단계에서는 이 NDJSON을 어떻게 파이썬에서 스트리밍 방식으로 읽고 쓰는지 살펴보겠습니다.

⚡ 스트리밍 방식으로 데이터 읽고 쓰기

NDJSON 포맷의 가장 큰 장점은 스트리밍 처리가 가능하다는 점입니다.
즉, 파일 전체를 한 번에 메모리에 올리지 않고, 한 줄씩 데이터를 읽거나 쓸 수 있기 때문에 대용량 데이터 처리에 적합합니다.
이는 특히 로그 파일, 센서 데이터, 실시간 이벤트 처리와 같은 시나리오에서 큰 이점을 제공합니다.

파이썬에서는 기본 파일 입출력과 json 모듈을 활용하면 쉽게 NDJSON 스트리밍을 구현할 수 있습니다.
데이터를 한 줄씩 읽으면서 JSON 객체로 변환하거나, 반대로 JSON 객체를 한 줄 단위로 직렬화하여 저장할 수 있습니다.

import json

# NDJSON 쓰기
data = [
    {"id": 1, "message": "로그 A"},
    {"id": 2, "message": "로그 B"},
    {"id": 3, "message": "로그 C"}
]

with open("logs.ndjson", "w", encoding="utf-8") as f:
    for record in data:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

# NDJSON 읽기
with open("logs.ndjson", "r", encoding="utf-8") as f:
    for line in f:
        obj = json.loads(line)
        print(obj)

위 코드에서 보듯이, NDJSON 파일은 한 줄 = 하나의 JSON 객체라는 규칙을 따릅니다.
쓰기 단계에서는 json.dumps()로 직렬화 후 줄바꿈과 함께 저장하고, 읽을 때는 각 줄을 json.loads()로 파싱하면 됩니다.

또한, 스트리밍 방식은 데이터 중간에 오류가 있더라도 전체 프로세스가 중단되지 않고, 문제 있는 행만 따로 처리할 수 있다는 장점이 있습니다.
이는 바로 다음 단계에서 다룰 에러 행 분리 저장 기법과도 밀접하게 연결됩니다.

🛡️ 에러 행 분리 저장 기법

스트리밍 처리의 핵심은 잘못된 레코드 때문에 전체 파이프라인이 멈추지 않도록 하는 데 있습니다.
NDJSON 파일은 줄 단위로 독립성이 보장되므로, 파싱 오류가 발생한 행만 별도로 분리해 저장하면 안정적으로 처리가 가능합니다.
이 섹션에서는 정상 레코드와 에러 레코드를 서로 다른 파일에 기록하고, 후속 검증과 재처리가 가능하도록 메타 정보를 함께 남기는 실전 패턴을 소개합니다.

🧩 라인 단위 파싱과 유효성 검증 파이프라인

오류 분리는 try-except로 JSON 파싱 단계에서 1차 수행하고, 스키마 검증으로 2차 수행하는 방식이 가장 단단합니다.
파싱이 실패하면 즉시 에러 파일에 원본 라인과 함께 에러 유형을 기록합니다.
파싱에 성공했더라도 필수 키 누락, 타입 불일치, 값 범위 위반 같은 도메인 유효성 검사에서 탈락할 수 있으므로, 검증 함수를 별도로 두면 재사용성이 높아집니다.

import json
from typing import Any, Dict

def is_valid(record: Dict[str, Any]) -> bool:
    # 필수 키 확인
    required = {"id", "message"}
    if not required.issubset(record):
        return False
    # 타입과 값 검증
    if not isinstance(record["id"], int):
        return False
    if not isinstance(record["message"], str) or not record["message"].strip():
        return False
    return True

def process_ndjson(
    src_path: str,
    ok_path: str,
    err_path: str,
    encoding: str = "utf-8"
) -> None:
    with open(src_path, "r", encoding=encoding) as src, \
         open(ok_path, "w", encoding=encoding) as ok, \
         open(err_path, "w", encoding=encoding) as err:

        line_no = 0
        for raw in src:
            line_no += 1
            line = raw.rstrip("\n")
            if not line:
                continue  # 빈 줄 무시

            try:
                obj = json.loads(line)
            except json.JSONDecodeError as e:
                error_entry = {
                    "line_no": line_no,
                    "error": "JSONDecodeError",
                    "detail": str(e),
                    "raw": line
                }
                err.write(json.dumps(error_entry, ensure_ascii=False) + "\n")
                continue

            if is_valid(obj):
                ok.write(json.dumps(obj, ensure_ascii=False) + "\n")
            else:
                error_entry = {
                    "line_no": line_no,
                    "error": "ValidationError",
                    "detail": "schema/type/value check failed",
                    "raw": line
                }
                err.write(json.dumps(error_entry, ensure_ascii=False) + "\n")

if __name__ == "__main__":
    process_ndjson("input.ndjson", "clean.ndjson", "errors.ndjson")

에러 파일에는 원본 라인과 함께 line_no, error, detail 같은 메타 필드를 포함시키는 것이 좋습니다.
이렇게 하면 재처리 스크립트에서 문제 유형별로 빠르게 재시도하거나, 별도 대시보드에서 통계를 산출하는 데 유리합니다.

🗂️ 정상/에러 파일 운영 전략

🧪 재처리 루프와 품질 제어

• 🔁에러 유형별 핸들러를 만들고, 성공 시 clean.ndjson로 이동
• 🧯같은 레코드의 반복 실패는 카운팅 후 격리(Dead-letter 처리)
• 📈라인 번호/에러코드 기준으로 에러율을 시계열로 추적

💡 실무에서 응용할 수 있는 활용 팁

NDJSON 스트리밍 처리와 에러 분리 저장은 단순히 파일을 다루는 기술을 넘어서, 데이터 파이프라인 운영에 필수적인 요소입니다.
실무 환경에서는 데이터 규모가 커질수록 안정성과 가시성을 높이는 전략이 필요합니다.
이 섹션에서는 운영 환경에서 바로 활용할 수 있는 실질적인 팁을 정리했습니다.

🛠️ 로깅과 모니터링

에러 행을 따로 저장하는 것만으로는 충분하지 않습니다.
에러율, 레코드 수, 처리 속도 같은 주요 지표를 로깅하고 시각화해야 병목이나 장애를 빠르게 파악할 수 있습니다.
Prometheus, Grafana 같은 오픈소스 모니터링 도구를 적용하면 알림과 대시보드를 손쉽게 구축할 수 있습니다.

📦 압축 및 저장소 관리

NDJSON 파일은 줄 수가 많아질수록 크기가 커집니다.
이를 효율적으로 보관하려면 gzip 압축을 적용해 용량을 줄이고, 클라우드 스토리지(S3, GCS 등)와 연동하여 보관하는 것이 일반적입니다.
또한 저장 경로를 날짜 기반 디렉터리 구조로 관리하면 검색과 추적이 쉬워집니다.

🔗 빅데이터 및 분석 플랫폼 연동

NDJSON은 Elasticsearch, Spark, Hadoop 같은 빅데이터 시스템과 호환성이 높습니다.
정상 데이터는 즉시 인덱싱하거나 배치 처리로 넘기고, 에러 데이터는 별도의 큐에 넣어 수동 검증을 진행할 수 있습니다.
이렇게 설계하면 데이터 품질을 유지하면서도 분석 워크플로우를 중단 없이 이어갈 수 있습니다.

• 📊처리량, 에러율, 지연시간 같은 지표 모니터링 구축
• 🗄️NDJSON 파일은 압축 및 클라우드 저장으로 운영 비용 절감
• ⚙️빅데이터 플랫폼과 실시간/배치 파이프라인 연동

❓ 자주 묻는 질문 (FAQ)