Processor JSONL Methods

Processor provides complete JSONL (JSON Lines) stream processing capabilities, supporting line-by-line processing, parallel processing, batch processing, and functional operations.

Stream Read Methods

StreamJSONL

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

Stream-processes JSONL data, reading line by line and calling the callback function.

Parameters

Name	Type	Description
reader	io.Reader	JSONL data source
fn	func(lineNum int, item *IterableValue) error	Processing function, returning an error stops processing

processor, _ := json.New()
defer processor.Close()

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

err := processor.StreamJSONL(file, func(lineNum int, item *json.IterableValue) error {
    level := item.GetString("level")
    msg := item.GetString("message")
    fmt.Printf("[%d] %s: %s\n", lineNum, level, msg)
    return nil
})

---

StreamJSONLParallel

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

Parallel-processes JSONL data using multiple worker goroutines to accelerate processing.

Parameters

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

processor, _ := json.New()
defer processor.Close()

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

var count int64
err := processor.StreamJSONLParallel(file, 8, func(lineNum int, item *json.IterableValue) error {
    atomic.AddInt64(&count, 1)
    // CPU-intensive processing...
    return nil
})
fmt.Printf("Processed %d lines\n", count)

Performance Tips

• Suitable for CPU-intensive operations (data transformation, computation)
• For I/O-intensive operations, use single-threaded StreamJSONL
• Workers count should be set to the number of CPU cores

StreamJSONLParallelWithContext

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

Parallel-processes JSONL data with context, supporting cancellation and timeout control.

Parameters

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

processor, _ := json.New()
defer processor.Close()

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

err := processor.StreamJSONLParallelWithContext(ctx, reader, 8, func(lineNum int, item *json.IterableValue) error {
    return nil
})
if err != nil {
    log.Fatal(err)
}

---

StreamJSONLChunked

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

Chunk-processes JSONL data, processing a batch of elements each time.

Parameters

Name	Type	Description
reader	io.Reader	JSONL data source
chunkSize	int	Number of elements per batch
fn	func(chunk []*IterableValue) error	Batch processing function

processor, _ := json.New()
defer processor.Close()

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

err := processor.StreamJSONLChunked(file, 100, func(chunk []*json.IterableValue) error {
    // Batch insert into database
    records := make([]Record, len(chunk))
    for i, item := range chunk {
        records[i] = Record{
            ID:    item.GetInt("id"),
            Name:  item.GetString("name"),
        }
    }
    return db.BatchInsert(records)
})

---

StreamJSONLFile

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

Stream-processes JSONL data directly from a file.

processor, _ := json.New()
defer processor.Close()

err := processor.StreamJSONLFile("logs.jsonl", func(lineNum int, item *json.IterableValue) error {
    if item.GetString("level") == "error" {
        logErrors(item)
    }
    return nil
})

---

Functional Operation Methods

ForeachJSONL

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

An alias method for iterating JSONL data, behaving identically to StreamJSONL.

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

---

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, returning the converted slice.

processor, _ := json.New()
defer processor.Close()

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

// Extract all usernames
names, err := processor.MapJSONL(file, func(lineNum int, item *json.IterableValue) (any, error) {
    return item.GetString("name"), nil
})
// names: []any{"Alice", "Bob", "Charlie"}

---

ReduceJSONL

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

Reduces JSONL data to a single value.

processor, _ := json.New()
defer processor.Close()

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

// Calculate total sales
total, err := processor.ReduceJSONL(file, 0.0, func(acc any, item *json.IterableValue) any {
    price := item.GetFloat64("price")
    return acc.(float64) + price
})
fmt.Printf("Total sales: %.2f\n", total.(float64))

---

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.

processor, _ := json.New()
defer processor.Close()

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

// Filter error logs
errors, err := processor.FilterJSONL(file, func(item *json.IterableValue) bool {
    return item.GetString("level") == "error"
})
fmt.Printf("Found %d error log entries\n", len(errors))

---

CollectJSONL

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

Collects all JSONL data into a slice.

processor, _ := json.New()
defer processor.Close()

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

items, err := processor.CollectJSONL(file)
if err != nil {
    panic(err)
}
fmt.Printf("Collected %d records\n", len(items))

Memory Note

This method loads all data into memory and is not suitable for very large files. For large files, use StreamJSONL for line-by-line processing.

---

FirstJSONL

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

Finds the first element that satisfies the condition.

Return Values

Type	Description
*IterableValue	Found element (if exists)
bool	Whether found
error	Error information

processor, _ := json.New()
defer processor.Close()

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

// Find the first admin
admin, found, err := processor.FirstJSONL(file, func(item *json.IterableValue) bool {
    return item.GetBool("is_admin")
})
if err != nil {
    panic(err)
}
if found {
    fmt.Printf("Admin: %s\n", admin.GetString("name"))
}

---

Configuration Options

JSONL processing behavior can be configured through the following Config fields:

Field	Type	Default	Description
JSONLBufferSize	int	65536 (64KB)	Read buffer size
JSONLMaxLineSize	int	1048576 (1MB)	Maximum bytes per line
JSONLSkipEmpty	bool	true	Skip empty lines
JSONLSkipComments	bool	false	Skip # or // comments
JSONLContinueOnErr	bool	false	Continue on parse error
JSONLWorkers	int	4	Parallel worker goroutines
JSONLChunkSize	int	1000	Batch size for chunk processing
JSONLMaxMemory	int64	104857600 (100MB)	Maximum memory usage

cfg := json.DefaultConfig()
cfg.JSONLSkipComments = true     // Skip comment lines
cfg.JSONLContinueOnErr = true    // Continue on parse error
cfg.JSONLWorkers = 8             // 8 parallel workers

processor, _ := json.New(cfg)
defer processor.Close()

---

Complete Examples

Log Analysis

package main

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

func main() {
    processor, _ := json.New()
    defer processor.Close()

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

    var errorCount, warningCount int

    err := processor.StreamJSONL(file, func(lineNum int, item *json.IterableValue) error {
        level := item.GetString("level")
        switch level {
        case "error":
            errorCount++
            fmt.Printf("[ERROR] %s\n", item.GetString("message"))
        case "warning":
            warningCount++
        }
        return nil
    })

    if err != nil {
        panic(err)
    }

    fmt.Printf("Statistics: %d errors, %d warnings\n", errorCount, warningCount)
}

Parallel Data Processing

package main

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

func main() {
    cfg := json.DefaultConfig()
    cfg.JSONLWorkers = 16 // 16 parallel workers

    processor, _ := json.New(cfg)
    defer processor.Close()

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

    var processed int64

    err := processor.StreamJSONLParallel(file, 16, func(lineNum int, item *json.IterableValue) error {
        // CPU-intensive processing (replace with your business logic)
        _ = item
        atomic.AddInt64(&processed, 1)
        return nil
    })

    if err != nil {
        panic(err)
    }

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

---

See Also

• JSONL Processor - Package-level JSONL functions
• Large File Processing - Large file processing guide
• Iterator - IterableValue type details