JSONL 데이터 파이프라인

이 문서는 CyberGo JSON으로 JSONL(줄바꿈 구분 JSON) 데이터 파이프라인을 구축하는 방법을 보여줍니다: 스트리밍 읽기, 필드 변환, 일괄 형식 변환, 대용량 파일 처리.

JSONL 스트리밍 읽기 및 변환

제네릭 StreamLinesInto[T]로 JSONL 스트림을 한 줄씩 읽어 구조체로 역직렬화하고, 콜백에서 필드를 변환한 뒤 ToJSONLString으로 JSONL 형식에 다시 씁니다.

package main

import (
	"fmt"
	"strings"

	"github.com/cybergodev/json"
)

// LogEntry는 단일 JSON 로그 줄을 나타냅니다
type LogEntry struct {
	Timestamp string `json:"timestamp"`
	Level     string `json:"level"`
	Message   string `json:"message"`
}

// EnrichedLog는 변환된 로그입니다 (필드 이름 변경 및 새 카테고리 추가)
type EnrichedLog struct {
	Timestamp string `json:"ts"`
	Level     string `json:"level"`
	Message   string `json:"msg"`
	Category  string `json:"category"`
}

func main() {
	// JSONL 로그 스트림 시뮬레이션 (실제로는 파일이나 네트워크에서 올 수 있음)
	jsonlStream := `{"timestamp":"2024-01-01T10:00:00Z","level":"INFO","message":"서비스 시작"}
{"timestamp":"2024-01-01T10:00:05Z","level":"ERROR","message":"데이터베이스 연결 실패"}
{"timestamp":"2024-01-01T10:00:10Z","level":"WARN","message":"응답 시간 임계값 초과"}
{"timestamp":"2024-01-01T10:00:15Z","level":"INFO","message":"재연결 성공"}`

	reader := strings.NewReader(jsonlStream)

	// 1. 각 로그 줄을 스트리밍 읽기 및 변환
	var enriched []any
	entries, err := json.StreamLinesInto[LogEntry](reader, func(lineNum int, entry LogEntry) error {
		// 레벨별 분류
		category := "normal"
		if entry.Level == "ERROR" {
			category = "critical"
		} else if entry.Level == "WARN" {
			category = "warning"
		}

		enriched = append(enriched, EnrichedLog{
			Timestamp: entry.Timestamp,
			Level:     entry.Level,
			Message:   entry.Message,
			Category:  category,
		})
		return nil
	})
	if err != nil {
		panic(err)
	}

	// 2. JSONL 형식으로 일괄 변환
	output, err := json.ToJSONLString(enriched)
	if err != nil {
		panic(err)
	}
	fmt.Printf("%d줄의 로그를 처리했습니다\n", len(entries))
	fmt.Print(output)
}

// 출력:
// 4줄의 로그를 처리했습니다
// {"ts":"2024-01-01T10:00:00Z","level":"INFO","msg":"서비스 시작","category":"normal"}
// {"ts":"2024-01-01T10:00:05Z","level":"ERROR","msg":"데이터베이스 연결 실패","category":"critical"}
// {"ts":"2024-01-01T10:00:10Z","level":"WARN","msg":"응답 시간 임계값 초과","category":"warning"}
// {"ts":"2024-01-01T10:00:15Z","level":"INFO","msg":"재연결 성공","category":"normal"}

JSONL 파일 처리

NDJSONProcessor는 JSONL 파일을 한 줄씩 처리하며, 콜백은 map[string]any를 받습니다 (필드가 고정되지 않은 경우에 유용). 집계 결과는 ToJSONL로 일괄 변환하여 JSONL 바이트로 만듭니다.

package main

import (
	"fmt"
	"os"
	"path/filepath"

	"github.com/cybergodev/json"
)

func main() {
	// 예제가 단독 실행되도록 임시 JSONL 파일 생성
	tmpDir, err := os.MkdirTemp("", "cybergo-pipeline-*")
	if err != nil {
		panic(err)
	}
	defer os.RemoveAll(tmpDir)

	jsonlPath := filepath.Join(tmpDir, "events.jsonl")
	jsonData := `{"event":"login","user":"alice","ts":"2024-01-01T10:00:00Z"}
{"event":"logout","user":"alice","ts":"2024-01-01T11:00:00Z"}
{"event":"login","user":"bob","ts":"2024-01-01T12:00:00Z"}
{"event":"purchase","user":"bob","ts":"2024-01-01T12:30:00Z"}`
	if err := os.WriteFile(jsonlPath, []byte(jsonData), 0644); err != nil {
		panic(err)
	}

	// 1. NDJSONProcessor로 줄별 처리 (각 줄은 map[string]any로 파싱)
	processor := json.NewNDJSONProcessor()
	loginCount := 0
	err = processor.ProcessFile(jsonlPath, func(lineNum int, obj map[string]any) error {
		event, _ := obj["event"].(string)
		user, _ := obj["user"].(string)
		fmt.Printf("%d번 줄: %s by %s\n", lineNum, event, user)
		if event == "login" {
			loginCount++
		}
		return nil
	})
	if err != nil {
		panic(err)
	}

	// 2. 집계 결과를 JSONL로 변환 (일괄 형식 변환)
	summary := []any{
		map[string]any{"metric": "logins", "count": loginCount},
		map[string]any{"metric": "total_events", "count": 4},
	}
	jsonlBytes, err := json.ToJSONL(summary)
	if err != nil {
		panic(err)
	}
	fmt.Printf("로그인 이벤트 수: %d\n", loginCount)
	fmt.Printf("집계 결과:\n%s", string(jsonlBytes))
}

// 출력:
// 1번 줄: login by alice
// 2번 줄: logout by alice
// 3번 줄: login by bob
// 4번 줄: purchase by bob
// 로그인 이벤트 수: 2
// 집계 결과:
// {"metric":"logins","count":2}
// {"metric":"total_events","count":4}

병렬 파이프라인: StreamJSONLParallel + JSONLWriter 쓰기

행 수가 많고 개별 행 처리가 무거울 때 (변환, 검증, 인리치먼트) StreamJSONLParallel 은 여러 worker 로 스트림을 병렬 소비합니다; 결과를 원본 행 순서대로 수집한 뒤 JSONLWriter.WriteRaw 로 재인코딩 없이 JSONL 로 씁니다:

package main

import (
	"bytes"
	"fmt"
	"slices"
	"strings"
	"sync"

	"github.com/cybergodev/json"
)

func main() {
	// 이벤트 로그 스트림 시뮬레이션 (실제로는 대용량 파일이 올 수 있음, strings.NewReader 를 os.Open 의 *os.File 로 교체)
	jsonlStream := `{"event":"login","user":"alice","ts":"10:00"}
{"event":"page_view","user":"alice","ts":"10:01"}
{"event":"login","user":"bob","ts":"10:02"}
{"event":"purchase","user":"bob","ts":"10:03"}
{"event":"login","user":"carol","ts":"10:04"}`

	p, err := json.New()
	if err != nil {
		panic(err)
	}
	defer p.Close()

	// 1. 병렬 필터 및 변환: login 이벤트만 유지하고 {user, at} 구조로 다시 씀.
	//    콜백은 여러 worker 에서 동시에 실행됨: 공유 map 쓰기에는 락 필요; lineNum 을 키로 저장하고 완료 후 원래 순서 복원
	var mu sync.Mutex
	logins := make(map[int][]byte)

	err = p.StreamJSONLParallel(strings.NewReader(jsonlStream), 4, func(lineNum int, item *json.IterableValue) error {
		if item.GetString("event") != "login" {
			return nil // 대상이 아닌 이벤트는 건너뜀; item.Break() 를 반환하면 전체 스트림을 깔끔하게 중단
		}
		encoded, err := json.Marshal(map[string]any{
			"user": item.GetString("user"),
			"at":   item.GetString("ts"),
		})
		if err != nil {
			return err // 오류를 반환하면 디스패치가 중단되고 오류가 그대로 보고됨
		}
		mu.Lock()
		logins[lineNum] = encoded
		mu.Unlock()
		return nil
	})
	if err != nil {
		panic(err)
	}

	// 2. 원본 행 순서대로 결과 출력 (WriteRaw 는 인코딩된 행을 그대로 쓰고 줄바꿈만 추가)
	lineNums := make([]int, 0, len(logins))
	for n := range logins {
		lineNums = append(lineNums, n)
	}
	slices.Sort(lineNums)

	var out bytes.Buffer
	writer := json.NewJSONLWriter(&out)
	for _, n := range lineNums {
		if err := writer.WriteRaw(logins[n]); err != nil {
			panic(err)
		}
	}

	fmt.Printf("로그인 이벤트 %d건을 걸러냈습니다 (%d줄 출력)\n", len(logins), writer.Stats().LinesProcessed)
	fmt.Print(out.String())
}

// 출력:
// 로그인 이벤트 3건을 걸러냈습니다 (3줄 출력)
// {"at":"10:00","user":"alice"}
// {"at":"10:02","user":"bob"}
// {"at":"10:04","user":"carol"}

병렬 파이프라인 핵심

순서: 병렬 콜백의 실행 순서는 보장되지 않지만 lineNum 은 항상 원본 행 번호에 대응합니다 — 행 번호로 수집한 뒤 정렬해 출력하면 순서가 유지됩니다.
worker 수: 두 번째 인수로 명시적으로 지정합니다 (예제는 4); 타임아웃/취소가 필요하면 StreamJSONLParallelWithContext(ctx, reader, workers, fn) 을 사용하세요.
처리량: 직렬 StreamJSONL 대비 이득은 개별 행 처리 오버헤드에 달려 있습니다 — 단순 추출형 가벼운 콜백은 향상이 제한적이고, 인리치먼트/검증형 무거운 콜백은 향상이 뚜렷합니다.

대용량 JSON 배열 파일 스트리밍

단일 대형 JSON 배열 파일(JSONL이 아님)의 경우, ForeachFile로 전체 파일을 메모리에 한 번에 로드하지 않고 요소별로 순회합니다.

package main

import (
	"fmt"
	"os"
	"path/filepath"

	"github.com/cybergodev/json"
)

func main() {
	tmpDir, err := os.MkdirTemp("", "cybergo-big-*")
	if err != nil {
		panic(err)
	}
	defer os.RemoveAll(tmpDir)

	// 대형 JSON 배열 파일 생성 (대규모 데이터셋 시뮬레이션)
	arrayPath := filepath.Join(tmpDir, "records.json")
	records := []any{
		map[string]any{"id": 1, "amount": 100, "currency": "USD"},
		map[string]any{"id": 2, "amount": 250, "currency": "EUR"},
		map[string]any{"id": 3, "amount": 80, "currency": "USD"},
		map[string]any{"id": 4, "amount": 500, "currency": "GBP"},
		map[string]any{"id": 5, "amount": 120, "currency": "USD"},
	}
	if err := json.SaveToFile(arrayPath, records); err != nil {
		panic(err)
	}

	// ForeachFile로 배열의 각 요소를 스트리밍 순회
	p, err := json.New()
	if err != nil {
		panic(err)
	}
	defer p.Close()

	totalUSD := 0
	err = p.ForeachFile(arrayPath, func(key any, item *json.IterableValue) error {
		currency := item.GetString("currency")
		amount := item.GetInt("amount")
		if currency == "USD" {
			totalUSD += amount
		}
		return nil // item.Break()를 반환하면 조기 중단
	})
	if err != nil {
		panic(err)
	}
	fmt.Printf("USD 총액: %d\n", totalUSD)
}

// 출력: USD 총액: 320

팁

JSONL 파일(줄마다 독립적인 JSON 객체 하나): StreamLinesInto[T], NDJSONProcessor, StreamJSONLFile을 사용하세요.
대형 JSON 배열 파일(단일 JSON 배열에 많은 요소): ForeachFile로 메모리에 모두 로드하지 않고 스트리밍하세요.

다음 단계

JSONL 스트리밍 — 전체 JSONL 처리 가이드
대용량 파일 처리 — 대용량 파일 스트리밍 상세
기본 예제 — 기본 JSONL 읽기/쓰기
치트시트 — 빠른 API 참조

JSONL 데이터 파이프라인 ​

JSONL 스트리밍 읽기 및 변환 ​

JSONL 파일 처리 ​

병렬 파이프라인: StreamJSONLParallel + JSONLWriter 쓰기 ​

대용량 JSON 배열 파일 스트리밍 ​

다음 단계 ​