대용량 JSON 다루기: 성능 최적화와 실용 기법

# 대용량 JSON 다루기: 성능 최적화와 실용 기법

대용량 JSON 파일은 특별한 처리가 필요합니다. 메모리 오버플로우 없이 효율적으로 처리하는 모든 방법을 알아보세요.

대용량 JSON의 문제점

일반적인 문제

크기별 분류

• 작음 (< 1MB): 문제 없음
• 중간 (1-10MB): 주의 필요
• 큼 (10-100MB): 특별 처리 필요
• 매우 큼 (> 100MB): 스트리밍 필수

스트리밍 파싱

Node.js - JSONStream

대용량 JSON을 스트림으로 처리:

npm install JSONStream

const fs = require('fs');
const JSONStream = require('JSONStream');

// 대용량 배열 처리
const stream = fs.createReadStream('large-data.json');
const parser = JSONStream.parse('items.');


let count = 0;

stream.pipe(parser);

parser.on('data', (item) => {
  // 각 항목을 개별적으로 처리
  console.log(item);
  count++;

  // 특정 조건으로 필터링
  if (item.value > 100) {
    // 처리 로직
  }
});

parser.on('end', () => {
  console.log(총 ${count}개 항목 처리 완료);
});

parser.on('error', (error) => {
  console.error('파싱 오류:', error);
});

const JSONStream = require('JSONStream');
const fs = require('fs');

// users 배열만 파싱
const parser = JSONStream.parse('users.');

fs.createReadStream('data.json')
  .pipe(parser)
  .on('data', (user) => {
    console.log('사용자:', user.name);
  });

쓰기 스트림

const JSONStream = require('JSONStream');
const fs = require('fs');

const writeStream = fs.createWriteStream('output.json');
const stringify = JSONStream.stringify();

stringify.pipe(writeStream);

// 데이터 쓰기
for (let i = 0; i < 1000000; i++) {
  stringify.write({
    id: i,
    name: User${i},
    timestamp: new Date().toISOString()
  });
}

stringify.end();

writeStream.on('finish', () => {
  console.log('쓰기 완료');
});

청크 단위 처리

Python - ijson

pip install ijson

import ijson

# 스트리밍 파싱
with open('large-data.json', 'rb') as f:
    # 'items' 배열의 각 항목 파싱
    parser = ijson.items(f, 'items.item')

    for item in parser:
        print(f"처리 중: {item['id']}")

        # 특정 조건으로 필터링
        if item['value'] > 100:
            # 처리 로직
            pass

print("완료")

특정 필드만 추출

import ijson

results = []

with open('large-data.json', 'rb') as f:
    # 이름 필드만 추출
    for name in ijson.items(f, 'users.item.name'):
        results.append(name)

print(f"총 {len(results)}개 이름 추출")

메모리 효율적 집계

import ijson

total = 0
count = 0

with open('large-data.json', 'rb') as f:
    for value in ijson.items(f, 'items.item.price'):
        total += value
        count += 1

average = total / count if count > 0 else 0
print(f"평균: {average}")

JSON Lines (JSONL)

JSONL 포맷

각 줄이 독립적인 JSON 객체:

{"id":1,"name":"홍길동","age":30}
{"id":2,"name":"김철수","age":25}
{"id":3,"name":"이영희","age":28}

Node.js에서 JSONL 처리

const fs = require('fs');
const readline = require('readline');

const rl = readline.createInterface({
  input: fs.createReadStream('data.jsonl'),
  crlfDelay: Infinity
});

let count = 0;

rl.on('line', (line) => {
  try {
    const data = JSON.parse(line);
    console.log('처리:', data.name);
    count++;
  } catch (error) {
    console.error('파싱 오류:', error);
  }
});

rl.on('close', () => {
  console.log(총 ${count}줄 처리);
});

Python에서 JSONL 처리

import json

results = []

with open('data.jsonl', 'r', encoding='utf-8') as f:
    for line in f:
        try:
            data = json.loads(line.strip())
            results.append(data)
        except json.JSONDecodeError as e:
            print(f'파싱 오류: {e}')

print(f'총 {len(results)}개 항목 처리')

Pandas로 JSONL 읽기

import pandas as pd

# 전체 로드
df = pd.read_json('data.jsonl', lines=True)

# 청크로 읽기
chunk_size = 10000

for chunk in pd.read_json('data.jsonl', lines=True, chunksize=chunk_size):
    # 각 청크 처리
    print(f'청크 크기: {len(chunk)}')
    # 처리 로직

메모리 최적화

필요한 필드만 로드

const JSONStream = require('JSONStream');
const fs = require('fs');

const parser = JSONStream.parse('users.');

const results = [];

fs.createReadStream('users.json')
  .pipe(parser)
  .on('data', (user) => {
    // 필요한 필드만 저장
    results.push({
      id: user.id,
      name: user.name
      // 다른 필드는 제외
    });
  });

import json

selected_fields = ['id', 'name', 'email']
results = []

with open('data.json', 'r') as f:
    data = json.load(f)

    for item in data:
        # 필요한 필드만 추출
        filtered = {k: item[k] for k in selected_fields if k in item}
        results.append(filtered)

import pandas as pd

# dtype 지정으로 메모리 절약
df = pd.read_json(
    'data.json',
    dtype={
        'id': 'int32',  # int64 대신
        'category': 'category',  # 범주형
        'value': 'float32',  # float64 대신
        'active': 'bool'
    }
)

print(df.memory_usage(deep=True))

const { Worker } = require('worker_threads');
const fs = require('fs');

// worker.js
if (require.main !== module) {
  const { parentPort } = require('worker_threads');

  parentPort.on('message', (chunk) => {
    // 청크 처리
    const result = chunk.map(item => ({
      ...item,
      processed: true
    }));

    parentPort.postMessage(result);
  });
}

// main.js
function processInParallel(data, workerCount = 4) {
  return new Promise((resolve) => {
    const chunkSize = Math.ceil(data.length / workerCount);
    const workers = [];
    const results = [];
    let completed = 0;

    for (let i = 0; i < workerCount; i++) {
      const start = i  chunkSize;
      const end = start + chunkSize;
      const chunk = data.slice(start, end);

      const worker = new Worker(__filename);
      workers.push(worker);

      worker.postMessage(chunk);

      worker.on('message', (result) => {
        results[i] = result;
        completed++;

        if (completed === workerCount) {
          resolve(results.flat());
        }
      });
    }
  });
}

Python - multiprocessing

import json
import multiprocessing as mp
from functools import partial

def process_chunk(chunk):
    # 청크 처리 로직
    return [item for item in chunk if item['value'] > 100]

def parallel_process(filename, num_workers=4):
    # 파일 읽기
    with open(filename, 'r') as f:
        data = json.load(f)

    # 청크로 분할
    chunk_size = len(data) // num_workers
    chunks = [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)]

    # 병렬 처리
    with mp.Pool(num_workers) as pool:
        results = pool.map(process_chunk, chunks)

    # 결과 합치기
    return [item for chunk in results for item in chunk]

if __name__ == '__main__':
    results = parallel_process('large-data.json')
    print(f'총 {len(results)}개 항목 처리')

압축

gzip으로 압축

const fs = require('fs');
const zlib = require('zlib');

// 압축
const input = fs.createReadStream('large-data.json');
const output = fs.createWriteStream('large-data.json.gz');
input.pipe(zlib.createGzip()).pipe(output);

// 압축 해제 및 파싱
const JSONStream = require('JSONStream');

fs.createReadStream('large-data.json.gz')
  .pipe(zlib.createGunzip())
  .pipe(JSONStream.parse('items.'))

  .on('data', (item) => {
    console.log(item);
  });

import gzip
import json

# 압축
with open('large-data.json', 'rb') as f_in:
    with gzip.open('large-data.json.gz', 'wb') as f_out:
        f_out.writelines(f_in)

# 압축 해제 및 파싱
with gzip.open('large-data.json.gz', 'rt', encoding='utf-8') as f:
    data = json.load(f)

const fs = require('fs');
const JSONStream = require('JSONStream');
const { MongoClient } = require('mongodb');

async function importToMongoDB() {
  const client = await MongoClient.connect('mongodb://localhost:27017');
  const db = client.db('mydb');
  const collection = db.collection('items');

  const stream = fs.createReadStream('large-data.json');
  const parser = JSONStream.parse('items.');

  let batch = [];
  const batchSize = 1000;

  stream.pipe(parser);

  parser.on('data', async (item) => {
    batch.push(item);

    if (batch.length >= batchSize) {
      await collection.insertMany(batch);
      console.log(${batch.length}개 삽입);
      batch = [];
    }
  });

  parser.on('end', async () => {
    if (batch.length > 0) {
      await collection.insertMany(batch);
    }
    await client.close();
    console.log('완료');
  });
}

importToMongoDB();

SQLite로 변환

import json
import sqlite3

conn = sqlite3.connect('data.db')
cursor = conn.cursor()

# 테이블 생성
cursor.execute('''
    CREATE TABLE IF NOT EXISTS items (
        id INTEGER PRIMARY KEY,
        name TEXT,
        value REAL
    )
''')

# 대용량 JSON 삽입
with open('large-data.json', 'r') as f:
    data = json.load(f)

    # 배치 삽입
    batch_size = 1000
    for i in range(0, len(data), batch_size):
        batch = data[i:i+batch_size]
        cursor.executemany(
            'INSERT INTO items VALUES (?, ?, ?)',
            [(item['id'], item['name'], item['value']) for item in batch]
        )
        conn.commit()
        print(f'{i + len(batch)}개 삽입')

conn.close()

파일 분할

큰 파일을 여러 개로 분할

const fs = require('fs');
const JSONStream = require('JSONStream');

const input = fs.createReadStream('large-data.json');
const parser = JSONStream.parse('items.');


let fileIndex = 0;
let itemCount = 0;
const itemsPerFile = 10000;
let currentBatch = [];

input.pipe(parser);

parser.on('data', (item) => {
  currentBatch.push(item);
  itemCount++;

  if (currentBatch.length >= itemsPerFile) {
    fs.writeFileSync(
      chunk_${fileIndex}.json,
      JSON.stringify(currentBatch, null, 2)
    );
    console.log(파일 ${fileIndex} 저장 (${currentBatch.length}개 항목));

    currentBatch = [];
    fileIndex++;
  }
});

parser.on('end', () => {
  if (currentBatch.length > 0) {
    fs.writeFileSync(
      chunk_${fileIndex}.json,
      JSON.stringify(currentBatch, null, 2)
    );
  }
  console.log(총 ${itemCount}개 항목을 ${fileIndex + 1}개 파일로 분할);
});

// main.js
const worker = new Worker('json-worker.js');

fetch('large-data.json')
  .then(res => res.text())
  .then(text => {
    worker.postMessage({ type: 'parse', data: text });
  });

worker.onmessage = (e) => {
  if (e.data.type === 'progress') {
    console.log(진행률: ${e.data.percent}%);
  } else if (e.data.type === 'complete') {
    console.log('완료:', e.data.result);
  }
};

// json-worker.js
self.onmessage = (e) => {
  if (e.data.type === 'parse') {
    const data = JSON.parse(e.data.data);

    // 청크로 처리
    const chunkSize = 1000;
    const results = [];

    for (let i = 0; i < data.length; i += chunkSize) {
      const chunk = data.slice(i, i + chunkSize);
      results.push(...chunk.filter(item => item.value > 100));

      // 진행률 보고
      self.postMessage({
        type: 'progress',
        percent: Math.round((i / data.length)  100)
      });
    }

    self.postMessage({
      type: 'complete',
      result: results
    });
  }
};

IndexedDB 사용

async function storeInIndexedDB(data) {
  const db = await openDB('mydb', 1, {
    upgrade(db) {
      db.createObjectStore('items', { keyPath: 'id' });
    }
  });

  const tx = db.transaction('items', 'readwrite');
  const store = tx.objectStore('items');

  // 배치 삽입
  for (const item of data) {
    await store.put(item);
  }

  await tx.done;
}

성능 모니터링

메모리 사용량 추적

function logMemoryUsage() {
  const usage = process.memoryUsage();
  console.log({
    rss: ${Math.round(usage.rss / 1024 / 1024)} MB,
    heapUsed: ${Math.round(usage.heapUsed / 1024 / 1024)} MB,
    external: ${Math.round(usage.external / 1024 / 1024)} MB
  });
}

setInterval(logMemoryUsage, 1000);

import psutil
import os

process = psutil.Process(os.getpid())

def log_memory():
    mem = process.memory_info()
    print(f'메모리 사용: {mem.rss / 1024 / 1024:.2f} MB')

# 정기적으로 호출

모범 사례

1. 적절한 도구 선택

• < 10MB: 일반 파싱
• 10-100MB: 청크 처리
• > 100MB: 스트리밍
• > 1GB: 데이터베이스

2. JSONL 사용

대용량 데이터는 JSONL 형식 권장:

{"id":1,"data":"..."}
{"id":2,"data":"..."}

3. 압축

항상 gzip 압축 고려

4. 인덱싱

자주 검색하는 데이터는 데이터베이스 사용

5. 모니터링

메모리와 처리 시간 추적

결론

대용량 JSON은 올바른 도구와 기법으로 효율적으로 처리할 수 있습니다. 스트리밍, 청크 처리, 병렬화를 적절히 활용하세요.

• ✅ 스트리밍으로 메모리 절약
• ✅ JSONL 형식 사용
• ✅ 필요한 필드만 로드
• ✅ 압축 활용
• ✅ 데이터베이스 고려

지금 바로 JSON Simplify에서 JSON을 최적화해보세요!