대용량 페이로드를 위한 JSON 파싱: 속도, 메모리 및 확장성의 균형

I’m ready to translate the article for you, but it looks like the text you’d like translated wasn’t included in your message. Could you please paste the content you want translated (excluding the source line you already provided)? Once I have the text, I’ll translate it into Korean while preserving all formatting, markdown, and code blocks.

Introduction

Black Friday 마케팅 캠페인이 대성공을 거두어 고객이 웹사이트에 몰려든다고 상상해 보세요. 평소에는 시간당 약 1 000개의 고객 이벤트만 처리하던 Mixpanel 설정이 같은 시간 안에 수백만 개의 이벤트를 받게 됩니다. 그 결과 데이터 파이프라인은 방대한 양의 JSON 데이터를 파싱하고 데이터베이스에 저장해야 하는 상황에 직면합니다.

표준 JSON 파싱 라이브러리는 급증한 데이터 양을 따라가지 못하고, 거의 실시간에 가까운 분석 보고서는 뒤처지게 됩니다. 바로 이때 효율적인 JSON 파싱 라이브러리의 중요성을 깨닫게 됩니다. 대용량 페이로드를 처리할 뿐만 아니라, 좋은 라이브러리는 깊게 중첩된 JSON 구조도 직렬화와 역직렬화를 원활히 수행해야 합니다.

이 글에서는 대용량 페이로드를 위한 Python 파싱 라이브러리를 살펴봅니다. ujson, orjson, ijson의 기능을 중점적으로 검토하고, 표준 라이브러리(json), ujson, orjson을 대상으로 직렬화와 역직렬화 성능을 벤치마크합니다.

Serialization = Python 객체를 JSON 문자열로 변환하는 과정.
Deserialization = JSON 문자열로부터 Python 객체를 재구성하는 과정.

후에 제시되는 의사결정 흐름도(diagram)를 통해 워크플로에 맞는 파서를 선택할 수 있습니다. 또한 NDJSON과 NDJSON 페이로드를 파싱할 수 있는 라이브러리도 다룹니다. 시작해 봅시다.

Stdlib json

표준 라이브러리는 모든 기본 Python 데이터 타입(dict, list, tuple 등)의 직렬화를 지원합니다. json.loads()를 호출하면 전체 JSON 문서가 한 번에 메모리로 로드됩니다. 이는 작은 페이로드에서는 문제없이 동작하지만, 큰 페이로드에서는 다음과 같은 문제를 일으킬 수 있습니다:

• 메모리 부족 오류
• 하위 워크플로우의 병목 현상

import json

with open("large_payload.json", "r") as f:
    json_data = json.loads(f)   # loads entire file into memory, all tokens at once

ijson

수백 메가바이트 규모의 페이로드에 대해 ijson(iterative json의 약자)은 파일을 한 번에 하나의 토큰씩 읽어 전체 문서를 메모리에 로드하는 오버헤드를 피합니다.

import ijson

with open("json_data.json", "r") as f:
    # 배열에서 한 번에 하나의 dict를 가져옵니다
    for record in ijson.items(f, "items.item"):
        process(record)   # ijson 라이브러리는 레코드를 토큰 단위로 읽어들입니다

따라서 ijson은 각 요소를 스트리밍하면서 Python dict로 변환하고 이를 여러분의 처리 함수(process(record))에 전달합니다.

ujson

ujson는 C 기반 구현에 Python 바인딩을 제공하기 때문에 순수 Python json 모듈보다 훨씬 빠른 대용량 JSON 페이로드 처리에 오랫동안 인기를 끌어왔습니다.

Note: 유지 관리자는 ujson을 maintenance‑only 모드로 전환했으므로, 새로운 프로젝트에서는 일반적으로 orjson을 선호합니다.

import ujson

taxonomy_data = (
    '{"id":1, "genus":"Thylacinus", "species":"cynocephalus", "extinct": true}'
)

# Deserialize
data_dict = ujson.loads(taxonomy_data)

# Serialize
with open("taxonomy_data.json", "w") as fh:
    ujson.dump(data_dict, fh)

# Deserialize again
with open("taxonomy_data.json", "r") as fh:
    data = ujson.load(fh)
    print(data)

orjson

orjson은 Rust로 작성되어 C 기반 라이브러리(ujson 등)와 달리 속도와 메모리 안전성을 보장합니다. 또한 dataclass와 datetime 같은 추가 Python 타입 직렬화를 지원합니다.

핵심 차이점: orjson.dumps()는 bytes를 반환하고, 다른 라이브러리들은 문자열을 반환합니다. 바이트를 반환하면 추가 인코딩 단계가 없어져 orjson의 높은 처리량에 기여합니다.

import json
import orjson

# Example payload
book_payload = (
    '{"id":1,"name":"The Great Gatsby","author":"F. Scott Fitzgerald"}'
)

# Serialize to bytes
json_bytes = orjson.dumps(json.loads(book_payload))

# Deserialize back to a Python object
obj = orjson.loads(json_bytes)
print(obj)

Decision Flow Diagram

Below is a simplified flow to help you pick the right parser:

               +-------------------+
               |  Payload size?    |
               +--------+----------+
                        |
          +-------------+-------------+
          |                           |
    100 MB)** – stream with `ijson`.  

100 MB 이하 – ijson으로 스트리밍

NDJSON (Newline‑Delimited JSON)

로그 스타일 데이터 처리 시, NDJSON이 더 적합한 경우가 많습니다. 각 줄이 유효한 JSON 문서이기 때문입니다. NDJSON을 파싱하는 방법:

• Standard json – 줄 단위로 읽기.
• orjson – 빠른 줄 단위 역직렬화 (orjson.loads(line)).
• ijson – 역시 동작하지만, 줄 단위 접근이 보통 더 간단합니다.

import orjson

with open("events.ndjson", "r") as f:
    for line in f:
        event = orjson.loads(line)
        process(event)

요약

대부분의 새로운 프로젝트에 대해:

• 메모리에 적합한 페이로드일 때 속도와 추가 타입 지원을 위해 **orjson**을 사용하세요.
• 정말 거대한 페이로드이거나 데이터를 점진적으로 처리해야 할 경우 **ijson**으로 전환하세요.

즐거운 파싱 되세요!

json, ujson, orjson 를 이용한 JSON 파싱 및 직렬화

import json
import ujson
import orjson

# Sample JSON payload
book_payload = '{"Title":"The Great Gatsby","Author":"F. Scott Fitzgerald","Publishing House":"Charles Scribner\'s Sons"}'

# Deserialize with orjson
data_dict = orjson.loads(book_payload)
print(data_dict)

# Serialize to a file
with open("book_data.json", "wb") as f:
    f.write(orjson.dumps(data_dict))   # Returns a bytes object

# Deserialize from the file
with open("book_data.json", "rb") as f:
    book_data = orjson.loads(f.read())
    print(book_data)

json, ujson, orjson 의 직렬화 기능 테스트

정수, 문자열, datetime 값을 포함하는 샘플 dataclass 객체를 생성합니다.

from dataclasses import dataclass
from datetime import datetime

@dataclass
class User:
    id: int
    name: str
    created: datetime

u = User(id=1, name="Thomas", created=datetime.now())

1. 표준 라이브러리 json

import json

try:
    print("json:", json.dumps(u))
except TypeError as e:
    print("json error:", e)

결과: json 은 dataclass 인스턴스나 datetime 객체를 직렬화할 수 없기 때문에 TypeError 를 발생시킵니다.

2. ujson

import ujson

try:
    print("ujson:", ujson.dumps(u))
except TypeError as e:
    print("ujson error:", e)

결과: ujson 도 dataclass와 datetime 값을 직렬화하지 못하고 실패합니다.

3. orjson

import orjson

try:
    print("orjson:", orjson.dumps(u))
except TypeError as e:
    print("orjson error:", e)

결과: orjson 은 dataclass와 datetime 객체를 모두 성공적으로 직렬화합니다.

NDJSON (줄 구분 JSON) 작업하기

NDJSON은 각 줄이 별도의 JSON 객체인 형식으로, 예를 들면 다음과 같습니다:

{"id": "A13434", "name": "Ella"}
{"id": "A13455", "name": "Charmont"}
{"id": "B32434", "name": "Areida"}

주로 로그와 스트리밍 데이터에 사용됩니다. 아래는 Python에서 NDJSON을 처리하는 세 가지 방법입니다.

표준 라이브러리 json을 이용한 NDJSON

import json

ndjson_payload = """{"id": "A13434", "name": "Ella"}
{"id": "A13455", "name": "Charmont"}
{"id": "B32434", "name": "Areida"}"""

# 페이로드를 파일에 기록
with open("json_lib.ndjson", "w", encoding="utf-8") as fh:
    for line in ndjson_payload.splitlines():
        fh.write(line.strip() + "\n")

# 한 줄씩 읽고 처리
with open("json_lib.ndjson", "r", encoding="utf-8") as fh:
    for line in fh:
        if line.strip():                     # 빈 줄은 건너뛰기
            item = json.loads(line)          # 역직렬화
            print(item)                      # 혹은 호출자 함수에 전달

ijson (스트리밍 파서)를 이용한 NDJSON

import ijson

ndjson_payload = """{"id": "A13434", "name": "Ella"}
{"id": "A13455", "name": "Charmont"}
{"id": "B32434", "name": "Areida"}"""

# 페이로드를 파일에 기록
with open("ijson_lib.ndjson", "w", encoding="utf-8") as fh:
    fh.write(ndjson_payload)

# 반복적으로 파싱
with open("ijson_lib.ndjson", "r", encoding="utf-8") as fh:
    for item in ijson.items(fh, "", multiple_values=True):
        print(item)

설명: ijson.items(fh, "", multiple_values=True)는 각 루트 요소(각 줄)를 별도의 JSON 객체로 취급하고 하나씩 반환합니다.

전용 ndjson 라이브러리를 이용한 NDJSON

import ndjson

ndjson_payload = """{"id": "A13434", "name": "Ella"}
{"id": "A13455", "name": "Charmont"}
{"id": "B32434", "name": "Areida"}"""

# 페이로드를 파일에 기록
with open("ndjson_lib.ndjson", "w", encoding="utf-8") as fh:
    fh.write(ndjson_payload)

# 파일 로드 – 딕셔너리 리스트를 반환
with open("ndjson_lib.ndjson", "r", encoding="utf-8") as fh:
    ndjson_data = ndjson.load(fh)
    print(ndjson_data)

핵심 정리

• 작은 규모에서 중간 규모의 NDJSON 페이로드는 표준 json 모듈을 사용해 한 줄씩 읽으면 충분합니다.
• 매우 큰 페이로드의 경우, ijson이 데이터를 스트리밍하고 메모리 사용을 최소화하므로 최선의 선택입니다.
• Python 객체에서 NDJSON을 생성해야 할 경우, ndjson 라이브러리가 편리합니다 (ndjson.dumps()가 변환을 자동으로 처리합니다).

왜 ijson이 벤치마킹에 포함되지 않았는가

ijson은 스트리밍 파서로, 우리가 벤치마크한 대용량 파서(json, ujson, orjson)와 근본적으로 다릅니다. 스트리밍 파서를 대용량 파서와 비교하는 것은 “사과와 오렌지를 비교하는” 것과 같습니다:

• 대용량 파서는 전체 JSON 문서를 메모리에 로드하고 속도를 최적화합니다.
• **ijson**은 문서를 점진적으로 처리하여 메모리 효율성을 최적화합니다.

속도만을 기준으로 한 벤치마크에 ijson을 포함하면, 메모리 사용량이 큰 JSON 스트림에 대한 주요 장점인 낮은 메모리 소비를 무시하고 가장 느린 것으로 잘못 표시될 수 있습니다. 따라서 ijson은 메모리 사용량이 주요 고려 사항일 때 별도로 평가됩니다.

벤치마킹을 위한 합성 JSON 페이로드 생성

우리는 mimesis 라이브러리를 사용해 1 백만 개 레코드를 포함하는 대용량 합성 JSON 페이로드를 생성합니다. 이 데이터는 JSON 라이브러리의 성능을 측정하는 데 활용될 수 있습니다. 아래 코드는 페이로드를 생성하며, 결과 파일은 대략 100 – 150 MB 정도로, 의미 있는 성능 테스트에 충분히 큰 크기입니다.

from mimesis import Person, Address
import json

person_name = Person("en")
complete_address = Address("en")

with open("large_payload.json", "w") as fh:   # 파일에 스트리밍
    fh.write("[")                           # JSON 배열 시작

    for i in range(1_000_000):
        payload = {
            "id": person_name.identifier(),
            "name": person_name.full_name(),
            "email": person_name.email(),
            "address": {
                "street": complete_address.street_name(),
                "city": complete_address.city(),
                "postal_code": complete_address.postal_code()
            }
        }

        json.dump(payload, fh)

        # 마지막 요소를 제외하고 모든 요소 뒤에 콤마 추가
        if i < 999_999:
            fh.write(",")

    fh.write("]")                           # JSON 배열 종료

샘플 출력

[
  {
    "id": "8177",
    "name": "Willia Hays",
    "email": "showers1819@yandex.com",
    "address": {
      "street": "Emerald Cove",
      "city": "Crown Point",
      "postal_code": "58293"
    }
  },
  {
    "id": "5931",
    "name": "Quinn Greer",
    "email": "professional2038@outlook.com",
    "address": {
      "street": "Ohlone",
      "city": "Bridgeport",
      "postal_code": "92982"
    }
  }
]

벤치마킹 시작하기

벤치마킹 전제 조건

JSON 파일을 문자열로 읽은 뒤 각 라이브러리의 loads() 함수를 사용해 역직렬화합니다.

with open("large_payload1.json", "r") as fh:
    payload_str = fh.read()   # raw JSON text

도우미 함수는 주어진 loads 구현을 세 번 실행하고 전체 경과 시간을 반환합니다.

import time

def benchmark_load(func, payload_str):
    start = time.perf_counter()
    for _ in range(3):
        func(payload_str)
    end = time.perf_counter()
    return end - start

역직렬화 속도 벤치마킹

import json, ujson, orjson

results = {
    "json.loads":  benchmark_load(json.loads,  payload_str),
    "ujson.loads": benchmark_load(ujson.loads, payload_str),
    "orjson.loads": benchmark_load(orjson.loads, payload_str),
}

for lib, t in results.items():
    print(f"{lib}: {t:.4f} seconds")

Result: orjson이 역직렬화에서 가장 빠릅니다.

직렬화 속도 벤치마킹

import json, ujson, orjson

def benchmark_dump(func, obj):
    start = time.perf_counter()
    for _ in range(3):
        func(obj)
    end = time.perf_counter()
    return end - start

# Example object (already loaded)
example_obj = json.loads(payload_str)

ser_results = {
    "json.dumps":  benchmark_dump(json.dumps,  example_obj),
    "ujson.dumps": benchmark_dump(ujson.dumps, example_obj),
    "orjson.dumps": benchmark_dump(orjson.dumps, example_obj),
}

for lib, t in ser_results.items():
    print(f"{lib}: {t:.4f} seconds")