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}

대용량 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 파일 처리 ​

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

다음 단계 ​

JSONL 데이터 파이프라인

JSONL 스트리밍 읽기 및 변환

JSONL 파일 처리

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

다음 단계