JSONL Data Pipeline

This guide demonstrates how to build JSONL (newline-delimited JSON) data pipelines with CyberGo JSON: streaming reads, field transformation, batch format conversion, and large-file processing.

Stream-Read and Transform JSONL

Use the generic StreamLinesInto[T] to read a JSONL stream line by line and deserialize into a struct, transform fields in the callback, then write back to JSONL format with ToJSONLString.

package main

import (
    "fmt"
    "strings"

    "github.com/cybergodev/json"
)

// LogEntry represents a single JSON log line
type LogEntry struct {
    Timestamp string `json:"timestamp"`
    Level     string `json:"level"`
    Message   string `json:"message"`
}

// EnrichedLog is the transformed log (renamed fields plus a new category)
type EnrichedLog struct {
    Timestamp string `json:"ts"`
    Level     string `json:"level"`
    Message   string `json:"msg"`
    Category  string `json:"category"`
}

func main() {
    // Simulate a JSONL log stream (could come from a file or network in practice)
    jsonlStream := `{"timestamp":"2024-01-01T10:00:00Z","level":"INFO","message":"service started"}
{"timestamp":"2024-01-01T10:00:05Z","level":"ERROR","message":"database connection failed"}
{"timestamp":"2024-01-01T10:00:10Z","level":"WARN","message":"response time exceeded threshold"}
{"timestamp":"2024-01-01T10:00:15Z","level":"INFO","message":"reconnected"}`

    reader := strings.NewReader(jsonlStream)

    // 1. Stream-read and transform each log line
    var enriched []any
    entries, err := json.StreamLinesInto[LogEntry](reader, func(lineNum int, entry LogEntry) error {
        // Categorize by level
        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. Batch-convert back to JSONL format
    output, err := json.ToJSONLString(enriched)
    if err != nil {
        panic(err)
    }
    fmt.Printf("processed %d log lines\n", len(entries))
    fmt.Print(output)
}
// Output:
// processed 4 log lines
// {"ts":"2024-01-01T10:00:00Z","level":"INFO","msg":"service started","category":"normal"}
// {"ts":"2024-01-01T10:00:05Z","level":"ERROR","msg":"database connection failed","category":"critical"}
// {"ts":"2024-01-01T10:00:10Z","level":"WARN","msg":"response time exceeded threshold","category":"warning"}
// {"ts":"2024-01-01T10:00:15Z","level":"INFO","msg":"reconnected","category":"normal"}

Processing JSONL Files

NDJSONProcessor processes a JSONL file line by line; the callback receives a map[string]any (handy when fields are not fixed). Aggregate results with ToJSONL to batch-convert to JSONL bytes.

package main

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

    "github.com/cybergodev/json"
)

func main() {
    // Create a temp JSONL file so the example runs standalone
    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. Process line by line with NDJSONProcessor (each line parsed into 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("Line %d: %s by %s\n", lineNum, event, user)
        if event == "login" {
            loginCount++
        }
        return nil
    })
    if err != nil {
        panic(err)
    }

    // 2. Convert aggregated results to JSONL (batch format conversion)
    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("Login events: %d\n", loginCount)
    fmt.Printf("Aggregated:\n%s", string(jsonlBytes))
}
// Output:
// Line 1: login by alice
// Line 2: logout by alice
// Line 3: login by bob
// Line 4: purchase by bob
// Login events: 2
// Aggregated:
// {"metric":"logins","count":2}
// {"metric":"total_events","count":4}

Streaming Large JSON Array Files

For a single large JSON array file (not JSONL), use ForeachFile to iterate element by element without loading the entire file into memory at once.

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)

    // Create a large JSON array file (simulating a big dataset)
    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)
    }

    // Stream over each element of the array with 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 // return item.Break() to stop early
    })
    if err != nil {
        panic(err)
    }
    fmt.Printf("Total USD: %d\n", totalUSD)
}
// Output: Total USD: 320

TIP

• JSONL files (one independent JSON object per line): use StreamLinesInto[T], NDJSONProcessor, or StreamJSONLFile.
• Large JSON array files (a single JSON array with many elements): use ForeachFile to stream without loading everything into memory.

Next Steps

• JSONL Streaming — full JSONL processing guide
• Large File Processing — streaming large files in depth
• Basic Examples — basic JSONL read/write usage
• Cheat Sheet — quick API reference