JSONL Processor

JSONL (JSON Lines) or NDJSON (Newline Delimited JSON) is a format with one JSON object per line. This library provides complete JSONL processing capabilities through Processor methods and package-level functions.

Format Specification

json

{"id":1,"name":"Alice"}
{"id":2,"name":"Bob"}
{"id":3,"name":"Charlie"}

Each line is a valid JSON value
Lines are separated by \n
The last line may or may not have a trailing newline

Processor JSONL Methods

JSONL processing functionality is provided through Processor methods.

StreamJSONL

Signature: func (p *Processor) StreamJSONL(reader io.Reader, fn func(lineNum int, item *IterableValue) error) error

Stream-processes JSONL data, returning one IterableValue per line.

Parameters

Name	Type	Description
`reader`	`io.Reader`	Data source
`fn`	`func(lineNum int, item *IterableValue) error`	Processing callback

Callback Return Values

Return Value	Description
`nil`	Continue to next line
`item.Break()`	Stop iteration without returning an error
other `error`	Stop iteration and return the error

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

file, _ := os.Open("data.jsonl")
defer file.Close()

err = p.StreamJSONL(file, func(lineNum int, item *json.IterableValue) error {
    name := item.GetString("name")
    age := item.GetInt("age")
    fmt.Printf("Line %d: name=%s, age=%d\n", lineNum, name, age)
    return nil // Continue processing
    // return item.Break() // Stop iteration
})

StreamJSONLParallel

Signature: func (p *Processor) StreamJSONLParallel(reader io.Reader, workers int, fn func(lineNum int, item *IterableValue) error) error

Parallel-processes JSONL data using a worker pool pattern.

Parameters

Name	Type	Description
`reader`	`io.Reader`	Data source
`workers`	`int`	Number of worker goroutines (defaults to 4 when <=0)
`fn`	`func(lineNum int, item *IterableValue) error`	Processing callback

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

err = p.StreamJSONLParallel(file, 8, func(lineNum int, item *json.IterableValue) error {
    // CPU-intensive processing
    return processItem(item)
})

Performance Tip

For CPU-intensive operations (such as data transformation and computation), parallel processing can significantly improve performance. For I/O-intensive operations, single-threaded processing is recommended.

StreamJSONLParallelWithContext

Signature: func (p *Processor) StreamJSONLParallelWithContext(ctx context.Context, reader io.Reader, workers int, fn func(lineNum int, item *IterableValue) error) error

Parallel processing of JSONL data with context. Supports timeout and cancellation.

Parameters

Name	Type	Description
`ctx`	`context.Context`	Context for cancellation and timeout
`reader`	`io.Reader`	Data source
`workers`	`int`	Number of worker goroutines (defaults to 4 when <=0)
`fn`	`func(lineNum int, item *IterableValue) error`	Processing callback

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

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

err = p.StreamJSONLParallelWithContext(ctx, file, 8, func(lineNum int, item *json.IterableValue) error {
    // Parallel processing with cancellation support
    return processItem(item)
})

StreamJSONLChunked

Signature: func (p *Processor) StreamJSONLChunked(reader io.Reader, chunkSize int, fn func(chunk []*IterableValue) error) error

Batch-processes JSONL data, processing the specified number of elements each time.

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

// 1000 per batch
err = p.StreamJSONLChunked(file, 1000, func(chunk []*json.IterableValue) error {
    // Batch insert into database
    for _, item := range chunk {
        processItem(item)
    }
    return nil
})

StreamJSONLFile

Signature: func (p *Processor) StreamJSONLFile(filename string, fn func(lineNum int, item *IterableValue) error) error

Directly processes a JSONL file.

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

err = p.StreamJSONLFile("data.jsonl", func(lineNum int, item *json.IterableValue) error {
    fmt.Printf("Line %d: %v\n", lineNum, item.GetData())
    return nil
})

Advanced JSONL Operations

MapJSONL

Signature: func (p *Processor) MapJSONL(reader io.Reader, fn func(lineNum int, item *IterableValue) (any, error)) ([]any, error)

Maps JSONL data to a new format.

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

result, err := p.MapJSONL(file, func(lineNum int, item *json.IterableValue) (any, error) {
    return map[string]any{
        "name": item.GetString("name"),
        "age":  item.GetInt("age"),
    }, nil
})

ReduceJSONL

Signature: func (p *Processor) ReduceJSONL(reader io.Reader, initial any, fn func(acc any, item *IterableValue) any) (any, error)

Aggregates JSONL data into a single result.

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

// Calculate total age
totalAge, err := p.ReduceJSONL(file, 0, func(acc any, item *json.IterableValue) any {
    return acc.(int) + item.GetInt("age")
})

FilterJSONL

Signature: func (p *Processor) FilterJSONL(reader io.Reader, predicate func(item *IterableValue) bool) ([]*IterableValue, error)

Filters JSONL data, returning elements that satisfy the condition.

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

// Filter adults
adults, err := p.FilterJSONL(file, func(item *json.IterableValue) bool {
    return item.GetInt("age") >= 18
})

CollectJSONL

Signature: func (p *Processor) CollectJSONL(reader io.Reader) ([]*IterableValue, error)

Collects all JSONL elements into a slice.

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

items, err := p.CollectJSONL(file)
for _, item := range items {
    fmt.Println(item.GetString("name"))
}

FirstJSONL

Signature: func (p *Processor) FirstJSONL(reader io.Reader, predicate func(item *IterableValue) bool) (*IterableValue, bool, error)

Returns the first element that satisfies the condition.

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

user, found, err := p.FirstJSONL(file, func(item *json.IterableValue) bool {
    return item.GetString("name") == "Alice"
})
if found {
    fmt.Println("Found:", user.GetString("name"))
}

ForeachJSONL

Signature: func (p *Processor) ForeachJSONL(reader io.Reader, fn func(lineNum int, item *IterableValue) error) error

Iterates JSONL data (alias for StreamJSONL).

JSONL Configuration

JSONL configuration is integrated into the Config struct:

cfg := json.DefaultConfig()
cfg.JSONLBufferSize = 128 * 1024    // Buffer size (default 64KB)
cfg.JSONLMaxLineSize = 2 * 1024 * 1024  // Maximum line size (default 1MB)
cfg.JSONLSkipEmpty = true           // Skip empty lines (default true)
cfg.JSONLSkipComments = true        // Skip comment lines (default false)
cfg.JSONLContinueOnErr = true       // Continue on parse error (default false)
cfg.JSONLWorkers = 8                // Parallel workers (default 4)
cfg.JSONLChunkSize = 500            // Chunk size (default 1000)
cfg.JSONLMaxMemory = 200 * 1024 * 1024 // Maximum memory (default 100MB)

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

JSONLWriter

The JSONL writer is used to write data in JSON Lines format.

NewJSONLWriter

Signature: func NewJSONLWriter(writer io.Writer, cfg ...Config) *JSONLWriter

Creates a JSONL writer. Supports optional configuration parameters.

file, _ := os.Create("output.jsonl")
defer file.Close()

// Use default configuration
writer := json.NewJSONLWriter(file)

// Use custom configuration
cfg := json.DefaultConfig()
cfg.EscapeHTML = true
writer = json.NewJSONLWriter(file, cfg)

Write

Signature: func (w *JSONLWriter) Write(data any) error

Writes a single JSON value as one line.

err := writer.Write(map[string]any{
    "id":   1,
    "name": "Alice",
})

WriteAll

Signature: func (w *JSONLWriter) WriteAll(data []any) error

Writes multiple JSON values, each as one line.

items := []any{
    map[string]any{"id": 1, "name": "Alice"},
    map[string]any{"id": 2, "name": "Bob"},
    map[string]any{"id": 3, "name": "Charlie"},
}

err := writer.WriteAll(items)

WriteRaw

Signature: func (w *JSONLWriter) WriteRaw(line []byte) error

Writes a raw JSON line (without JSON encoding).

err := writer.WriteRaw([]byte(`{"id":1,"name":"raw"}`))

Err

Signature: func (w *JSONLWriter) Err() error

Returns errors that occurred during writing.

if err := writer.Err(); err != nil {
    fmt.Printf("Write error: %v\n", err)
}

Stats

Signature: func (w *JSONLWriter) Stats() JSONLStats

Gets write statistics.

stats := writer.Stats()
fmt.Printf("Wrote %d lines, %d bytes\n", stats.LinesProcessed, stats.BytesWritten)

JSONLStats Structure:

type JSONLStats struct {
    LinesProcessed int64 // Number of lines processed
    BytesWritten   int64 // Number of bytes written
}

NDJSONProcessor

A specialized processor for map[string]any type NDJSON files.

NewNDJSONProcessor

Signature: func NewNDJSONProcessor(cfg ...Config) *NDJSONProcessor

Creates an NDJSON processor. Supports optional configuration parameters.

// Use default configuration
np := json.NewNDJSONProcessor()

// Use custom configuration
cfg := json.DefaultConfig()
cfg.JSONLBufferSize = 128 * 1024
np = json.NewNDJSONProcessor(cfg)

ProcessFile

Signature: func (np *NDJSONProcessor) ProcessFile(filename string, fn func(lineNum int, obj map[string]any) error) error

Processes an NDJSON file.

err := np.ProcessFile("data.ndjson", func(lineNum int, obj map[string]any) error {
    fmt.Printf("[%d] ID: %v\n", lineNum, obj["id"])
    return nil
})

ProcessReader

Signature: func (np *NDJSONProcessor) ProcessReader(reader io.Reader, fn func(lineNum int, obj map[string]any) error) error

Processes NDJSON from a Reader.

err := np.ProcessReader(file, func(lineNum int, obj map[string]any) error {
    return nil
})

Package-Level Functions

All JSONL processing functions provide package-level convenience versions with signatures identical to the corresponding Processor methods. They use the default global Processor internally, so no manual instance creation is required.

Tip

Package-level functions are suitable for one-off processing. If you need to call them repeatedly in a loop or share configuration, create a dedicated Processor (json.New()) to reuse the cache.

StreamJSONL

Signature: func StreamJSONL(reader io.Reader, fn func(lineNum int, item *IterableValue) error, cfg ...Config) error

Stream-processes JSONL line by line, parsing each line into an IterableValue before invoking the callback.

StreamJSONLParallel

Signature: func StreamJSONLParallel(reader io.Reader, workers int, fn func(lineNum int, item *IterableValue) error, cfg ...Config) error

Processes JSONL using workers parallel goroutines.

StreamJSONLParallelWithContext

Signature: func StreamJSONLParallelWithContext(ctx context.Context, reader io.Reader, workers int, fn func(lineNum int, item *IterableValue) error, cfg ...Config) error

Parallel JSONL processing with context cancellation support.

StreamJSONLChunked

Signature: func StreamJSONLChunked(reader io.Reader, chunkSize int, fn func(chunk []*IterableValue) error, cfg ...Config) error

Processes JSONL in batches of chunkSize, passing each batch as []*IterableValue to the callback.

ForeachJSONL

Signature: func ForeachJSONL(reader io.Reader, fn func(lineNum int, item *IterableValue) error, cfg ...Config) error

Iterates over JSONL, invoking the callback for each line.

MapJSONL

Signature: func MapJSONL(reader io.Reader, fn func(lineNum int, item *IterableValue) (any, error), cfg ...Config) ([]any, error)

Maps each line to a new value and returns a slice of results.

ReduceJSONL

Signature: func ReduceJSONL(reader io.Reader, initial any, fn func(acc any, item *IterableValue) any, cfg ...Config) (any, error)

Reduces JSONL; initial is the accumulator's initial value.

FilterJSONL

Signature: func FilterJSONL(reader io.Reader, predicate func(item *IterableValue) bool, cfg ...Config) ([]*IterableValue, error)

Filters JSONL by predicate, returning the matching items.

StreamJSONLFile

Signature: func StreamJSONLFile(filename string, fn func(lineNum int, item *IterableValue) error, cfg ...Config) error

Package-level function that stream-processes JSONL data directly from a file without creating a Processor.

err := json.StreamJSONLFile("data.jsonl", func(lineNum int, item *json.IterableValue) error {
    fmt.Printf("Line %d: %v\n", lineNum, item.GetData())
    return nil
})

CollectJSONL

Signature: func CollectJSONL(reader io.Reader, cfg ...Config) ([]*IterableValue, error)

Reads all JSONL lines and collects them into a slice.

FirstJSONL

Signature: func FirstJSONL(reader io.Reader, predicate func(item *IterableValue) bool, cfg ...Config) (*IterableValue, bool, error)

Returns the first element satisfying the predicate; the second return value indicates whether a match was found.

StreamLinesInto[T]

Signature: func StreamLinesInto[T any](reader io.Reader, fn func(lineNum int, data T) error, cfg ...Config) ([]T, error)

Stream-reads JSONL and processes line by line.

type User struct {
    ID   int    `json:"id"`
    Name string `json:"name"`
}

// Use default configuration
entries, err := json.StreamLinesInto[User](file, func(lineNum int, user User) error {
    fmt.Printf("Processing: %s\n", user.Name)
    return nil
})

// Use custom configuration
cfg := json.DefaultConfig()
cfg.JSONLSkipComments = true
entries, err = json.StreamLinesInto[User](file, func(lineNum int, user User) error {
    return nil
}, cfg)

ParseJSONL

Signature: func ParseJSONL(data []byte, cfg ...Config) ([]any, error)

Parses a JSONL byte slice.

jsonl := `{"name":"Alice"}
{"name":"Bob"}`
results, err := json.ParseJSONL([]byte(jsonl))

ToJSONL

Signature: func ToJSONL(data []any, cfg ...Config) ([]byte, error)

Converts to a JSONL byte slice.

items := []any{
    map[string]any{"id": 1},
    map[string]any{"id": 2},
}
jsonl, err := json.ToJSONL(items)

ToJSONLString

Signature: func ToJSONLString(data []any, cfg ...Config) (string, error)

Converts to a JSONL string.

jsonlStr, err := json.ToJSONLString(items)

Complete Examples

Reading a Large JSONL File

package main

import (
    "fmt"
    "os"
    "github.com/cybergodev/json"
)

type LogEntry struct {
    Time    string `json:"time"`
    Level   string `json:"level"`
    Message string `json:"message"`
}

func main() {
    file, _ := os.Open("logs.jsonl")
    defer file.Close()

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

    count := 0
    err = p.StreamJSONL(file, func(lineNum int, item *json.IterableValue) error {
        count++
        if item.GetString("level") == "error" {
            fmt.Printf("Error: %s\n", item.GetString("message"))
        }
        return nil
    })

    if err != nil {
        fmt.Printf("Error: %v\n", err)
    }

    fmt.Printf("Total processed %d lines\n", count)
}

Writing a JSONL File

package main

import (
    "fmt"
    "os"
    "github.com/cybergodev/json"
)

func main() {
    file, _ := os.Create("output.jsonl")
    defer file.Close()

    writer := json.NewJSONLWriter(file)

    for i := 0; i < 10; i++ {
        writer.Write(map[string]any{
            "id":    i,
            "value": fmt.Sprintf("item-%d", i),
        })
    }

    stats := writer.Stats()
    fmt.Printf("Wrote %d bytes\n", stats.BytesWritten)
}

Parallel Processing of Large Files

package main

import (
    "fmt"
    "os"
    "sync/atomic"
    "github.com/cybergodev/json"
)

func main() {
    file, _ := os.Open("large.jsonl")
    defer file.Close()

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

    var count int64
    err = p.StreamJSONLParallel(file, 8, func(lineNum int, item *json.IterableValue) error {
        atomic.AddInt64(&count, 1)
        return nil
    })

    if err != nil {
        panic(err)
    }

    fmt.Printf("Parallel processed %d lines\n", count)
}

JSONL Processor ​

Format Specification ​

Processor JSONL Methods ​

StreamJSONL ​

StreamJSONLParallel ​

StreamJSONLParallelWithContext ​

StreamJSONLChunked ​

StreamJSONLFile ​

Advanced JSONL Operations ​

MapJSONL ​

ReduceJSONL ​

FilterJSONL ​

CollectJSONL ​

FirstJSONL ​

ForeachJSONL ​

JSONL Configuration ​

JSONLWriter ​

NewJSONLWriter ​

Write ​

WriteAll ​

WriteRaw ​

Err ​

Stats ​

NDJSONProcessor ​

NewNDJSONProcessor ​

ProcessFile ​

ProcessReader ​

Package-Level Functions ​

StreamJSONL ​

StreamJSONLParallel ​

StreamJSONLParallelWithContext ​

StreamJSONLChunked ​

ForeachJSONL ​

MapJSONL ​

ReduceJSONL ​

FilterJSONL ​

StreamJSONLFile ​

CollectJSONL ​

FirstJSONL ​

StreamLinesInto[T] ​

ParseJSONL ​

ToJSONL ​

ToJSONLString ​

Complete Examples ​

Reading a Large JSONL File ​

Writing a JSONL File ​

Parallel Processing of Large Files ​

See Also ​

JSONL Processor

Format Specification

Processor JSONL Methods

StreamJSONL

StreamJSONLParallel

StreamJSONLParallelWithContext

StreamJSONLChunked

StreamJSONLFile

Advanced JSONL Operations

MapJSONL

ReduceJSONL

FilterJSONL

CollectJSONL

FirstJSONL

ForeachJSONL

JSONL Configuration

JSONLWriter

NewJSONLWriter

Write

WriteAll

WriteRaw

Err

Stats

NDJSONProcessor

NewNDJSONProcessor

ProcessFile

ProcessReader

Package-Level Functions

StreamJSONL

StreamJSONLParallel

StreamJSONLParallelWithContext

StreamJSONLChunked

ForeachJSONL

MapJSONL

ReduceJSONL

FilterJSONL

StreamJSONLFile

CollectJSONL

FirstJSONL

StreamLinesInto[T]

ParseJSONL

ToJSONL

ToJSONLString

Complete Examples

Reading a Large JSONL File

Writing a JSONL File

Parallel Processing of Large Files

See Also