JSONL Code - Python, JavaScript, Go & More

Jump to a Language

Python

Reading JSONL (Standard Library)

The simplest approach using Python's built-in json module:

import json

# Basic line-by-line reading
with open('data.jsonl', 'r', encoding='utf-8') as f:
    for line in f:
        # Skip empty lines
        line = line.strip()
        if not line:
            continue

        # Parse JSON
        record = json.loads(line)
        print(record['name'])

# With error handling
with open('data.jsonl', 'r', encoding='utf-8') as f:
    for line_num, line in enumerate(f, 1):
        try:
            record = json.loads(line)
            process(record)
        except json.JSONDecodeError as e:
            print(f"Error on line {line_num}: {e}")
            continue

Writing JSONL

import json

# Write list of dicts to JSONL
data = [
    {"id": 1, "name": "Alice", "age": 30},
    {"id": 2, "name": "Bob", "age": 25},
]

with open('output.jsonl', 'w', encoding='utf-8') as f:
    for record in data:
        # Write JSON + newline
        f.write(json.dumps(record) + '\n')

# Append to existing file
with open('output.jsonl', 'a', encoding='utf-8') as f:
    new_record = {"id": 3, "name": "Charlie", "age": 35}
    f.write(json.dumps(new_record) + '\n')

# Pretty-print with ensure_ascii=False for Unicode
with open('output.jsonl', 'w', encoding='utf-8') as f:
    for record in data:
        f.write(json.dumps(record, ensure_ascii=False) + '\n')

High-Performance Parsing (orjson)

orjson is 3-5x faster than standard library and handles more edge cases:

# Install: pip install orjson
import orjson

# Read JSONL (note: read as binary 'rb')
with open('data.jsonl', 'rb') as f:
    for line in f:
        record = orjson.loads(line)
        print(record['name'])

# Write JSONL (note: orjson.dumps returns bytes)
with open('output.jsonl', 'wb') as f:
    for record in data:
        # orjson.dumps() returns bytes, add b'\n'
        f.write(orjson.dumps(record) + b'\n')

# Pretty formatting option
with open('output.jsonl', 'wb') as f:
    for record in data:
        json_bytes = orjson.dumps(
            record,
            option=orjson.OPT_INDENT_2  # Pretty print (not typical for JSONL)
        )
        f.write(json_bytes + b'\n')

Performance tip: orjson is especially fast for large objects and handles datetime serialization automatically.

Working with Pandas

Pandas has native JSONL support via lines=True parameter:

import pandas as pd

# Read JSONL into DataFrame
df = pd.read_json('data.jsonl', lines=True)
print(df.head())

# Write DataFrame to JSONL
df.to_json('output.jsonl', orient='records', lines=True)

# Read compressed JSONL
df = pd.read_json('data.jsonl.gz', lines=True, compression='gzip')

# Chunked reading for large files (memory efficient)
chunk_size = 10000
chunks = []

for chunk in pd.read_json('large.jsonl', lines=True, chunksize=chunk_size):
    # Process each chunk
    filtered = chunk[chunk['age'] > 25]
    chunks.append(filtered)

# Combine all chunks
result = pd.concat(chunks, ignore_index=True)

# Stream processing without loading all into memory
with pd.read_json('large.jsonl', lines=True, chunksize=10000) as reader:
    for chunk in reader:
        # Process and write incrementally
        processed = transform(chunk)
        processed.to_json('output.jsonl', orient='records', lines=True, mode='a')

Streaming Large Files

Process files larger than available RAM:

import json

def stream_jsonl(filepath, batch_size=1000):
    """Generator that yields batches of records"""
    batch = []

    with open(filepath, 'r', encoding='utf-8') as f:
        for line in f:
            try:
                record = json.loads(line)
                batch.append(record)

                if len(batch) >= batch_size:
                    yield batch
                    batch = []
            except json.JSONDecodeError:
                continue

    # Yield remaining records
    if batch:
        yield batch

# Usage: process in batches
for batch in stream_jsonl('huge.jsonl', batch_size=5000):
    # Process batch (5000 records at a time)
    results = bulk_process(batch)
    save_results(results)

# Filter and transform streaming
def filter_transform_jsonl(input_path, output_path, condition):
    with open(input_path, 'r') as fin, open(output_path, 'w') as fout:
        for line in fin:
            record = json.loads(line)

            # Apply filter
            if condition(record):
                # Transform
                transformed = transform(record)
                fout.write(json.dumps(transformed) + '\n')

# Example: Extract active users
filter_transform_jsonl(
    'users.jsonl',
    'active_users.jsonl',
    lambda r: r.get('status') == 'active'
)

Working with Compressed JSONL

import gzip
import json

# Read gzipped JSONL (streaming)
with gzip.open('data.jsonl.gz', 'rt', encoding='utf-8') as f:
    for line in f:
        record = json.loads(line)
        process(record)

# Write gzipped JSONL
with gzip.open('output.jsonl.gz', 'wt', encoding='utf-8') as f:
    for record in data:
        f.write(json.dumps(record) + '\n')

# Transparent compression wrapper
def open_jsonl(filepath, mode='r'):
    """Open JSONL file, handling .gz automatically"""
    if filepath.endswith('.gz'):
        return gzip.open(filepath, mode + 't', encoding='utf-8')
    else:
        return open(filepath, mode, encoding='utf-8')

# Usage works for both compressed and uncompressed
with open_jsonl('data.jsonl.gz', 'r') as f:
    for line in f:
        record = json.loads(line)

# Read bzip2, xz, or zstd
import bz2
import lzma

# bzip2
with bz2.open('data.jsonl.bz2', 'rt') as f:
    for line in f:
        record = json.loads(line)

# xz/lzma
with lzma.open('data.jsonl.xz', 'rt') as f:
    for line in f:
        record = json.loads(line)

Parallel Processing

from multiprocessing import Pool
import json

def process_chunk(lines):
    """Process a chunk of lines"""
    results = []
    for line in lines:
        try:
            record = json.loads(line)
            result = expensive_computation(record)
            results.append(result)
        except:
            continue
    return results

def parallel_process_jsonl(filepath, num_workers=4, chunk_size=10000):
    """Process JSONL in parallel"""
    # Read file into chunks
    chunks = []
    current_chunk = []

    with open(filepath, 'r') as f:
        for line in f:
            current_chunk.append(line)

            if len(current_chunk) >= chunk_size:
                chunks.append(current_chunk)
                current_chunk = []

    if current_chunk:
        chunks.append(current_chunk)

    # Process chunks in parallel
    with Pool(processes=num_workers) as pool:
        all_results = pool.map(process_chunk, chunks)

    # Flatten results
    final_results = [item for sublist in all_results for item in sublist]
    return final_results

# Usage
results = parallel_process_jsonl('data.jsonl', num_workers=8)

# Alternative: concurrent.futures for more control
from concurrent.futures import ProcessPoolExecutor, as_completed

def process_file_parallel(filepath):
    # Split file into parts
    chunks = split_file_into_chunks(filepath, num_chunks=10)

    with ProcessPoolExecutor(max_workers=10) as executor:
        # Submit all chunks
        futures = {executor.submit(process_chunk_file, chunk): chunk
                   for chunk in chunks}

        # Collect results as they complete
        for future in as_completed(futures):
            chunk = futures[future]
            try:
                result = future.result()
                print(f"Chunk {chunk} completed")
            except Exception as e:
                print(f"Chunk {chunk} failed: {e}")

Streaming with ijson

ijson provides iterative JSON parsing for ultra-large files:

# Install: pip install ijson
import ijson

# Stream large JSONL files with minimal memory
with open('large.jsonl', 'rb') as f:
    # Parse each JSON object as it's encountered
    objects = ijson.items(f, '', multiple_values=True)

    for obj in objects:
        print(obj['name'])
        process(obj)

# Memory-efficient: only loads one object at a time
# Perfect for files larger than available RAM

# Advanced: extract specific fields only
with open('data.jsonl', 'rb') as f:
    # Only parse the 'name' field from each object
    names = ijson.items(f, 'item.name', multiple_values=True)
    for name in names:
        print(name)

# Combine with generators for pipelines
def extract_active_users(filepath):
    with open(filepath, 'rb') as f:
        objects = ijson.items(f, '', multiple_values=True)
        for obj in objects:
            if obj.get('status') == 'active':
                yield obj

# Use in pipeline
for user in extract_active_users('users.jsonl'):
    process(user)

Best for: Files too large to fit in memory, where you need true streaming parsing.

Using jsonlines Library

The jsonlines library provides a clean API for JSONL operations:

# Install: pip install jsonlines
import jsonlines

# Read JSONL
with jsonlines.open('data.jsonl') as reader:
    for record in reader:
        print(record['name'])

# Write JSONL
with jsonlines.open('output.jsonl', mode='w') as writer:
    writer.write({'name': 'Alice', 'age': 30})
    writer.write({'name': 'Bob', 'age': 25})
    # Or write multiple at once
    writer.write_all([
        {'name': 'Charlie', 'age': 35},
        {'name': 'Diana', 'age': 28}
    ])

# Read with compression
with jsonlines.open('data.jsonl.gz') as reader:
    for record in reader:
        process(record)

# Skip invalid lines
with jsonlines.open('data.jsonl', mode='r') as reader:
    for record in reader.iter(skip_invalid=True):
        process(record)

# Type checking with dataclasses
from dataclasses import dataclass

@dataclass
class User:
    id: int
    name: str
    age: int

with jsonlines.open('users.jsonl') as reader:
    for obj in reader:
        user = User(**obj)  # Convert to dataclass
        print(user.name)

Complete Python Example: ETL Pipeline

#!/usr/bin/env python3
"""
ETL Pipeline: Read JSONL, transform, filter, write output
Handles large files, errors, and compression
"""
import json
import gzip
import logging
from typing import Iterator, Dict, Any
from pathlib import Path

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def read_jsonl(filepath: str) -> Iterator[Dict[str, Any]]:
    """Read JSONL file (handles .gz compression)"""
    open_func = gzip.open if filepath.endswith('.gz') else open

    with open_func(filepath, 'rt', encoding='utf-8') as f:
        for line_num, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue

            try:
                yield json.loads(line)
            except json.JSONDecodeError as e:
                logger.error(f"Parse error line {line_num}: {e}")


def write_jsonl(records: Iterator[Dict[str, Any]], filepath: str):
    """Write records to JSONL file (handles .gz compression)"""
    open_func = gzip.open if filepath.endswith('.gz') else open

    with open_func(filepath, 'wt', encoding='utf-8') as f:
        for record in records:
            f.write(json.dumps(record) + '\n')


def transform(record: Dict[str, Any]) -> Dict[str, Any]:
    """Transform record (example: add computed field)"""
    record['full_name'] = f"{record['first_name']} {record['last_name']}"
    record['is_adult'] = record['age'] >= 18
    return record


def filter_record(record: Dict[str, Any]) -> bool:
    """Filter logic (example: active users only)"""
    return record.get('status') == 'active'


def etl_pipeline(input_file: str, output_file: str):
    """Complete ETL pipeline"""
    logger.info(f"Starting ETL: {input_file} -> {output_file}")

    processed = 0
    filtered = 0

    records = read_jsonl(input_file)
    transformed = (transform(r) for r in records)
    filtered_records = (r for r in transformed if filter_record(r))

    write_jsonl(filtered_records, output_file)

    logger.info(f"ETL complete. Processed: {processed}, Output: {filtered}")


if __name__ == '__main__':
    etl_pipeline('input.jsonl.gz', 'output.jsonl.gz')

JavaScript / Node.js

Reading JSONL (Node.js)

const fs = require('fs');
const readline = require('readline');

// Basic line-by-line reading
async function readJSONL(filepath) {
    const fileStream = fs.createReadStream(filepath);

    const rl = readline.createInterface({
        input: fileStream,
        crlfDelay: Infinity  // Handle both \n and \r\n
    });

    for await (const line of rl) {
        if (!line.trim()) continue;  // Skip empty lines

        try {
            const record = JSON.parse(line);
            console.log(record.name);
        } catch (err) {
            console.error('Parse error:', err.message);
        }
    }
}

// Usage
readJSONL('data.jsonl');

// Callback-based approach
function readJSONLCallback(filepath, onRecord, onComplete) {
    const fileStream = fs.createReadStream(filepath);
    const rl = readline.createInterface({ input: fileStream });

    rl.on('line', (line) => {
        try {
            const record = JSON.parse(line);
            onRecord(record);
        } catch (err) {
            console.error('Parse error:', err);
        }
    });

    rl.on('close', onComplete);
}

Writing JSONL

const fs = require('fs');

// Basic writing
const data = [
    { id: 1, name: 'Alice', age: 30 },
    { id: 2, name: 'Bob', age: 25 }
];

const writeStream = fs.createWriteStream('output.jsonl');

data.forEach(record => {
    writeStream.write(JSON.stringify(record) + '\n');
});

writeStream.end();

// Async/await with promises
async function writeJSONL(records, filepath) {
    const writeStream = fs.createWriteStream(filepath);

    for (const record of records) {
        const json = JSON.stringify(record) + '\n';

        // Wait if buffer is full
        if (!writeStream.write(json)) {
            await new Promise(resolve => writeStream.once('drain', resolve));
        }
    }

    writeStream.end();
    return new Promise(resolve => writeStream.once('finish', resolve));
}

// Usage
await writeJSONL(data, 'output.jsonl');

// Append to existing file
const appendStream = fs.createWriteStream('output.jsonl', { flags: 'a' });
appendStream.write(JSON.stringify({ id: 3, name: 'Charlie' }) + '\n');
appendStream.end();

High-Performance Streaming (ndjson)

The ndjson npm package provides optimized streaming:

// Install: npm install ndjson
const fs = require('fs');
const ndjson = require('ndjson');

// Read JSONL with streaming parser
fs.createReadStream('data.jsonl')
    .pipe(ndjson.parse())
    .on('data', (record) => {
        console.log(record.name);
    })
    .on('end', () => {
        console.log('Done');
    });

// Write JSONL with streaming serializer
const writeStream = fs.createWriteStream('output.jsonl');
const stringify = ndjson.stringify();

stringify.pipe(writeStream);

// Write objects
stringify.write({ id: 1, name: 'Alice' });
stringify.write({ id: 2, name: 'Bob' });
stringify.end();

// Transform pipeline
fs.createReadStream('input.jsonl')
    .pipe(ndjson.parse())
    .pipe(ndjson.stringify())
    .pipe(fs.createWriteStream('output.jsonl'));

Transform Streams

const { Transform } = require('stream');
const fs = require('fs');
const ndjson = require('ndjson');

// Custom transform stream
class FilterTransform extends Transform {
    constructor(filterFn) {
        super({ objectMode: true });
        this.filterFn = filterFn;
    }

    _transform(record, encoding, callback) {
        if (this.filterFn(record)) {
            // Transform record
            record.processed = true;
            record.timestamp = Date.now();
            this.push(record);
        }
        callback();
    }
}

// Usage: filter active users
const filterActive = new FilterTransform(r => r.status === 'active');

fs.createReadStream('users.jsonl')
    .pipe(ndjson.parse())
    .pipe(filterActive)
    .pipe(ndjson.stringify())
    .pipe(fs.createWriteStream('active_users.jsonl'));

// Pipeline with multiple transforms
const { pipeline } = require('stream/promises');

async function processJSONL() {
    await pipeline(
        fs.createReadStream('input.jsonl'),
        ndjson.parse(),
        new FilterTransform(r => r.age > 25),
        ndjson.stringify(),
        fs.createWriteStream('output.jsonl')
    );
    console.log('Pipeline complete');
}

processJSONL();

Compressed JSONL

const fs = require('fs');
const zlib = require('zlib');
const readline = require('readline');

// Read gzipped JSONL
async function readGzippedJSONL(filepath) {
    const fileStream = fs.createReadStream(filepath);
    const gunzip = zlib.createGunzip();
    const rl = readline.createInterface({
        input: fileStream.pipe(gunzip),
        crlfDelay: Infinity
    });

    for await (const line of rl) {
        const record = JSON.parse(line);
        console.log(record);
    }
}

// Write gzipped JSONL
async function writeGzippedJSONL(records, filepath) {
    const writeStream = fs.createWriteStream(filepath);
    const gzip = zlib.createGzip();

    gzip.pipe(writeStream);

    for (const record of records) {
        gzip.write(JSON.stringify(record) + '\n');
    }

    gzip.end();
    return new Promise(resolve => writeStream.on('finish', resolve));
}

// With ndjson
const ndjson = require('ndjson');

fs.createReadStream('data.jsonl.gz')
    .pipe(zlib.createGunzip())
    .pipe(ndjson.parse())
    .on('data', record => {
        console.log(record);
    });

HTTP Streaming API

const express = require('express');
const fs = require('fs');
const ndjson = require('ndjson');

const app = express();

// Streaming JSONL endpoint
app.get('/api/data', (req, res) => {
    res.setHeader('Content-Type', 'application/x-ndjson');
    res.setHeader('Transfer-Encoding', 'chunked');

    fs.createReadStream('data.jsonl')
        .pipe(ndjson.parse())
        .pipe(ndjson.stringify())
        .pipe(res);
});

// Query database and stream results
app.get('/api/users', async (req, res) => {
    res.setHeader('Content-Type', 'application/x-ndjson');

    const cursor = db.collection('users').find().stream();

    cursor.on('data', (doc) => {
        res.write(JSON.stringify(doc) + '\n');
    });

    cursor.on('end', () => {
        res.end();
    });
});

app.listen(3000);

// Client: consume streaming JSONL
const https = require('https');
const readline = require('readline');

https.get('https://api.example.com/data', (res) => {
    const rl = readline.createInterface({ input: res });

    rl.on('line', (line) => {
        const record = JSON.parse(line);
        console.log(record);
    });
});

Browser Usage (Fetch API)

// Streaming JSONL in the browser
async function fetchStreamingJSONL(url) {
    const response = await fetch(url);
    const reader = response.body.getReader();
    const decoder = new TextDecoder();

    let buffer = '';

    while (true) {
        const { done, value } = await reader.read();

        if (done) break;

        // Decode chunk
        buffer += decoder.decode(value, { stream: true });

        // Process complete lines
        const lines = buffer.split('\n');
        buffer = lines.pop(); // Keep incomplete line in buffer

        for (const line of lines) {
            if (line.trim()) {
                const record = JSON.parse(line);
                console.log(record);
                // Update UI with record
            }
        }
    }
}

// Usage
fetchStreamingJSONL('https://api.example.com/data.jsonl');

// With async iterator (modern browsers)
async function* parseJSONLStream(response) {
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';

    while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });
        const lines = buffer.split('\n');
        buffer = lines.pop();

        for (const line of lines) {
            if (line.trim()) {
                yield JSON.parse(line);
            }
        }
    }
}

// Usage with for-await
const response = await fetch('https://api.example.com/data.jsonl');
for await (const record of parseJSONLStream(response)) {
    console.log(record);
}

Complete Node.js Example: API + Streaming

// Complete example: JSONL API with filtering
const express = require('express');
const fs = require('fs');
const ndjson = require('ndjson');
const { Transform } = require('stream');

const app = express();

// Custom filter transform
class FilterStream extends Transform {
    constructor(query) {
        super({ objectMode: true });
        this.query = query;
    }

    _transform(record, encoding, callback) {
        // Apply filters
        let match = true;

        if (this.query.minAge && record.age < this.query.minAge) {
            match = false;
        }

        if (this.query.status && record.status !== this.query.status) {
            match = false;
        }

        if (match) {
            this.push(record);
        }

        callback();
    }
}

// Streaming JSONL endpoint with query params
app.get('/api/users', (req, res) => {
    const query = {
        minAge: parseInt(req.query.minAge) || 0,
        status: req.query.status
    };

    res.setHeader('Content-Type', 'application/x-ndjson');
    res.setHeader('Transfer-Encoding', 'chunked');

    const filterStream = new FilterStream(query);

    fs.createReadStream('users.jsonl')
        .pipe(ndjson.parse())
        .pipe(filterStream)
        .pipe(ndjson.stringify())
        .pipe(res)
        .on('error', (err) => {
            console.error('Stream error:', err);
            res.end();
        });
});

app.listen(3000, () => {
    console.log('API listening on port 3000');
});

Go

Reading JSONL (bufio.Scanner)

package main

import (
    "bufio"
    "encoding/json"
    "fmt"
    "log"
    "os"
)

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

func main() {
    file, err := os.Open("data.jsonl")
    if err != nil {
        log.Fatal(err)
    }
    defer file.Close()

    scanner := bufio.NewScanner(file)

    // Increase buffer size for large lines (default is 64KB)
    const maxCapacity = 1024 * 1024 // 1MB
    buf := make([]byte, maxCapacity)
    scanner.Buffer(buf, maxCapacity)

    lineNum := 0
    for scanner.Scan() {
        lineNum++
        line := scanner.Bytes()

        var user User
        if err := json.Unmarshal(line, &user); err != nil {
            log.Printf("Error on line %d: %v", lineNum, err)
            continue
        }

        fmt.Printf("%s\n", user.Name)
    }

    if err := scanner.Err(); err != nil {
        log.Fatal(err)
    }
}

Writing JSONL

package main

import (
    "bufio"
    "encoding/json"
    "log"
    "os"
)

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

func main() {
    users := []User{
        {ID: 1, Name: "Alice", Email: "[email protected]"},
        {ID: 2, Name: "Bob", Email: "[email protected]"},
    }

    file, err := os.Create("output.jsonl")
    if err != nil {
        log.Fatal(err)
    }
    defer file.Close()

    // Use buffered writer for better performance
    writer := bufio.NewWriter(file)
    defer writer.Flush()

    encoder := json.NewEncoder(writer)

    for _, user := range users {
        if err := encoder.Encode(&user); err != nil {
            log.Printf("Error encoding user %d: %v", user.ID, err)
            continue
        }
    }
}

// Append to existing file
func appendToJSONL(user User, filepath string) error {
    file, err := os.OpenFile(filepath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
    if err != nil {
        return err
    }
    defer file.Close()

    encoder := json.NewEncoder(file)
    return encoder.Encode(&user)
}

Streaming with json.Decoder

package main

import (
    "bufio"
    "encoding/json"
    "io"
    "log"
    "os"
)

type Record struct {
    ID   int    `json:"id"`
    Data string `json:"data"`
}

// Stream JSONL with json.Decoder (more efficient than Scanner for large files)
func streamJSONL(filepath string, processFn func(Record) error) error {
    file, err := os.Open(filepath)
    if err != nil {
        return err
    }
    defer file.Close()

    reader := bufio.NewReader(file)

    for {
        line, err := reader.ReadBytes('\n')
        if err == io.EOF {
            break
        }
        if err != nil {
            return err
        }

        var record Record
        if err := json.Unmarshal(line, &record); err != nil {
            log.Printf("Parse error: %v", err)
            continue
        }

        if err := processFn(record); err != nil {
            return err
        }
    }

    return nil
}

func main() {
    err := streamJSONL("data.jsonl", func(r Record) error {
        // Process each record
        log.Printf("Processing ID %d", r.ID)
        return nil
    })

    if err != nil {
        log.Fatal(err)
    }
}

Concurrent Processing with Goroutines

package main

import (
    "bufio"
    "encoding/json"
    "log"
    "os"
    "sync"
)

type Record struct {
    ID   int    `json:"id"`
    Data string `json:"data"`
}

// Concurrent JSONL processing
func processConcurrent(filepath string, numWorkers int) error {
    file, err := os.Open(filepath)
    if err != nil {
        return err
    }
    defer file.Close()

    // Channel for distributing work
    lines := make(chan []byte, numWorkers*2)
    results := make(chan Record, numWorkers*2)

    // Start worker goroutines
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for line := range lines {
                var record Record
                if err := json.Unmarshal(line, &record); err != nil {
                    log.Printf("Parse error: %v", err)
                    continue
                }

                // Process record
                processed := processRecord(record)
                results <- processed
            }
        }()
    }

    // Result collector
    var resultsWg sync.WaitGroup
    resultsWg.Add(1)
    go func() {
        defer resultsWg.Done()
        for result := range results {
            // Handle results
            log.Printf("Processed: %+v", result)
        }
    }()

    // Read file and distribute lines
    scanner := bufio.NewScanner(file)
    buf := make([]byte, 1024*1024)
    scanner.Buffer(buf, 1024*1024)

    for scanner.Scan() {
        // Make a copy of line (scanner reuses buffer)
        lineCopy := make([]byte, len(scanner.Bytes()))
        copy(lineCopy, scanner.Bytes())
        lines <- lineCopy
    }

    close(lines)
    wg.Wait()
    close(results)
    resultsWg.Wait()

    return scanner.Err()
}

func processRecord(r Record) Record {
    // Expensive processing here
    return r
}

func main() {
    if err := processConcurrent("data.jsonl", 10); err != nil {
        log.Fatal(err)
    }
}

Compressed JSONL

package main

import (
    "bufio"
    "compress/gzip"
    "encoding/json"
    "log"
    "os"
)

type Record struct {
    ID   int    `json:"id"`
    Data string `json:"data"`
}

// Read gzipped JSONL
func readGzippedJSONL(filepath string) error {
    file, err := os.Open(filepath)
    if err != nil {
        return err
    }
    defer file.Close()

    gzReader, err := gzip.NewReader(file)
    if err != nil {
        return err
    }
    defer gzReader.Close()

    scanner := bufio.NewScanner(gzReader)
    buf := make([]byte, 1024*1024)
    scanner.Buffer(buf, 1024*1024)

    for scanner.Scan() {
        var record Record
        if err := json.Unmarshal(scanner.Bytes(), &record); err != nil {
            log.Printf("Parse error: %v", err)
            continue
        }

        // Process record
        log.Printf("%+v", record)
    }

    return scanner.Err()
}

// Write gzipped JSONL
func writeGzippedJSONL(records []Record, filepath string) error {
    file, err := os.Create(filepath)
    if err != nil {
        return err
    }
    defer file.Close()

    gzWriter := gzip.NewWriter(file)
    defer gzWriter.Close()

    writer := bufio.NewWriter(gzWriter)
    defer writer.Flush()

    encoder := json.NewEncoder(writer)

    for _, record := range records {
        if err := encoder.Encode(&record); err != nil {
            return err
        }
    }

    return nil
}

func main() {
    // Read
    if err := readGzippedJSONL("data.jsonl.gz"); err != nil {
        log.Fatal(err)
    }

    // Write
    records := []Record{
        {ID: 1, Data: "test"},
    }
    if err := writeGzippedJSONL(records, "output.jsonl.gz"); err != nil {
        log.Fatal(err)
    }
}

Generic JSONL Processing

These helpers use type parameters, so they need a Go toolchain with generics support (Go 1.18 or newer). Any currently supported Go release qualifies.

package main

import (
    "bufio"
    "encoding/json"
    "io"
    "os"
)

// Generic JSONL reader
func ReadJSONL[T any](filepath string) ([]T, error) {
    file, err := os.Open(filepath)
    if err != nil {
        return nil, err
    }
    defer file.Close()

    var results []T
    scanner := bufio.NewScanner(file)

    for scanner.Scan() {
        var record T
        if err := json.Unmarshal(scanner.Bytes(), &record); err != nil {
            continue
        }
        results = append(results, record)
    }

    return results, scanner.Err()
}

// Generic JSONL writer
func WriteJSONL[T any](records []T, filepath string) error {
    file, err := os.Create(filepath)
    if err != nil {
        return err
    }
    defer file.Close()

    writer := bufio.NewWriter(file)
    defer writer.Flush()

    encoder := json.NewEncoder(writer)

    for _, record := range records {
        if err := encoder.Encode(&record); err != nil {
            return err
        }
    }

    return nil
}

// Streaming generic processor
func StreamJSONL[T any](r io.Reader, processFn func(T) error) error {
    scanner := bufio.NewScanner(r)

    for scanner.Scan() {
        var record T
        if err := json.Unmarshal(scanner.Bytes(), &record); err != nil {
            continue
        }

        if err := processFn(record); err != nil {
            return err
        }
    }

    return scanner.Err()
}

type User struct {
    Name string `json:"name"`
    Age  int    `json:"age"`
}

func main() {
    // Type-safe reading
    users, _ := ReadJSONL[User]("users.jsonl")

    // Type-safe writing
    WriteJSONL(users, "output.jsonl")

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

    StreamJSONL(file, func(u User) error {
        // Process user
        return nil
    })
}

Complete Go Example: Concurrent ETL Pipeline

package main

import (
    "bufio"
    "compress/gzip"
    "encoding/json"
    "log"
    "os"
    "sync"
)

type InputRecord struct {
    ID     int    `json:"id"`
    Status string `json:"status"`
    Value  int    `json:"value"`
}

type OutputRecord struct {
    ID        int    `json:"id"`
    Processed bool   `json:"processed"`
    Result    int    `json:"result"`
}

func main() {
    if err := runETL("input.jsonl.gz", "output.jsonl.gz", 10); err != nil {
        log.Fatal(err)
    }
}

func runETL(inputPath, outputPath string, numWorkers int) error {
    // Open input file
    inFile, err := os.Open(inputPath)
    if err != nil {
        return err
    }
    defer inFile.Close()

    gzReader, _ := gzip.NewReader(inFile)
    defer gzReader.Close()

    // Open output file
    outFile, err := os.Create(outputPath)
    if err != nil {
        return err
    }
    defer outFile.Close()

    gzWriter := gzip.NewWriter(outFile)
    defer gzWriter.Close()

    // Channels
    inputs := make(chan InputRecord, numWorkers*2)
    outputs := make(chan OutputRecord, numWorkers*2)

    // Workers
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for input := range inputs {
                // Filter
                if input.Status != "active" {
                    continue
                }

                // Transform
                output := OutputRecord{
                    ID:        input.ID,
                    Processed: true,
                    Result:    input.Value * 2,
                }

                outputs <- output
            }
        }()
    }

    // Writer goroutine
    var writerWg sync.WaitGroup
    writerWg.Add(1)
    go func() {
        defer writerWg.Done()
        encoder := json.NewEncoder(gzWriter)
        for output := range outputs {
            encoder.Encode(&output)
        }
    }()

    // Read and distribute
    scanner := bufio.NewScanner(gzReader)
    buf := make([]byte, 1024*1024)
    scanner.Buffer(buf, 1024*1024)

    for scanner.Scan() {
        var record InputRecord
        if err := json.Unmarshal(scanner.Bytes(), &record); err != nil {
            continue
        }
        inputs <- record
    }

    // Cleanup
    close(inputs)
    wg.Wait()
    close(outputs)
    writerWg.Wait()

    return scanner.Err()
}

Java

Setup and Dependencies

Add Jackson or Gson to your project for JSON processing:

// Maven - Jackson
<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
    <version>2.15.2</version>
</dependency>

// Gradle - Jackson
implementation 'com.fasterxml.jackson.core:jackson-databind:2.15.2'

// Maven - Gson
<dependency>
    <groupId>com.google.code.gson</groupId>
    <artifactId>gson</artifactId>
    <version>2.10.1</version>
</dependency>

// Gradle - Gson
implementation 'com.google.code.gson:gson:2.10.1'

Reading JSONL with Jackson

import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class JacksonJSONLReader {
    public static class User {
        public int id;
        public String name;
        public String email;
    }

    public static void main(String[] args) throws IOException {
        ObjectMapper mapper = new ObjectMapper();
        List<User> users = new ArrayList<>();

        try (BufferedReader reader = new BufferedReader(new FileReader("data.jsonl"))) {
            String line;
            int lineNum = 0;

            while ((line = reader.readLine()) != null) {
                lineNum++;

                // Skip empty lines
                if (line.trim().isEmpty()) {
                    continue;
                }

                try {
                    User user = mapper.readValue(line, User.class);
                    users.add(user);
                    System.out.println(user.name);
                } catch (IOException e) {
                    System.err.println("Error on line " + lineNum + ": " + e.getMessage());
                }
            }
        }

        System.out.println("Loaded " + users.size() + " users");
    }
}

Writing JSONL with Jackson

import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;

public class JacksonJSONLWriter {
    public static class User {
        public int id;
        public String name;
        public String email;

        public User(int id, String name, String email) {
            this.id = id;
            this.name = name;
            this.email = email;
        }
    }

    public static void main(String[] args) throws IOException {
        ObjectMapper mapper = new ObjectMapper();

        List<User> users = Arrays.asList(
            new User(1, "Alice", "[email protected]"),
            new User(2, "Bob", "[email protected]"),
            new User(3, "Charlie", "[email protected]")
        );

        try (BufferedWriter writer = new BufferedWriter(new FileWriter("output.jsonl"))) {
            for (User user : users) {
                String json = mapper.writeValueAsString(user);
                writer.write(json);
                writer.newLine();
            }
        }

        System.out.println("Wrote " + users.size() + " users to output.jsonl");
    }
}

Streaming Large Files with Jackson

Jackson's streaming API provides memory-efficient processing:

import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.function.Consumer;

public class StreamingJSONLProcessor {
    public static class Record {
        public int id;
        public String data;
    }

    public static void streamJSONL(String filepath, Consumer<Record> processor) throws IOException {
        ObjectMapper mapper = new ObjectMapper();

        try (BufferedReader reader = new BufferedReader(new FileReader(filepath), 8192)) {
            String line;

            while ((line = reader.readLine()) != null) {
                if (line.trim().isEmpty()) {
                    continue;
                }

                try {
                    Record record = mapper.readValue(line, Record.class);
                    processor.accept(record);
                } catch (IOException e) {
                    System.err.println("Parse error: " + e.getMessage());
                }
            }
        }
    }

    public static void main(String[] args) throws IOException {
        // Process each record as it's read (low memory usage)
        streamJSONL("large.jsonl", record -> {
            System.out.println("Processing: " + record.id);
            // Do expensive processing here
        });
    }
}

Reading JSONL with Gson

import com.google.gson.Gson;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class GsonJSONLReader {
    public static class User {
        int id;
        String name;
        String email;
    }

    public static void main(String[] args) throws IOException {
        Gson gson = new Gson();
        List<User> users = new ArrayList<>();

        try (BufferedReader reader = new BufferedReader(new FileReader("data.jsonl"))) {
            String line;

            while ((line = reader.readLine()) != null) {
                if (line.trim().isEmpty()) {
                    continue;
                }

                User user = gson.fromJson(line, User.class);
                users.add(user);
                System.out.println(user.name);
            }
        }

        System.out.println("Total users: " + users.size());
    }
}

Writing JSONL with Gson

import com.google.gson.Gson;
import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;

public class GsonJSONLWriter {
    public static class User {
        int id;
        String name;
        String email;

        User(int id, String name, String email) {
            this.id = id;
            this.name = name;
            this.email = email;
        }
    }

    public static void main(String[] args) throws IOException {
        Gson gson = new Gson();

        List<User> users = Arrays.asList(
            new User(1, "Alice", "[email protected]"),
            new User(2, "Bob", "[email protected]")
        );

        try (BufferedWriter writer = new BufferedWriter(new FileWriter("output.jsonl"))) {
            for (User user : users) {
                String json = gson.toJson(user);
                writer.write(json);
                writer.newLine();
            }
        }
    }
}

Parallel Processing with Streams

import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import java.util.stream.Collectors;

public class ParallelJSONLProcessor {
    public static class Record {
        public int id;
        public String data;
        public String status;
    }

    public static void main(String[] args) throws IOException {
        ObjectMapper mapper = new ObjectMapper();

        // Read and process in parallel
        List<Record> results = Files.lines(Paths.get("data.jsonl"))
            .parallel()
            .filter(line -> !line.trim().isEmpty())
            .map(line -> {
                try {
                    return mapper.readValue(line, Record.class);
                } catch (IOException e) {
                    return null;
                }
            })
            .filter(record -> record != null)
            .filter(record -> "active".equals(record.status))
            .map(record -> {
                // Expensive processing
                record.data = processData(record.data);
                return record;
            })
            .collect(Collectors.toList());

        System.out.println("Processed " + results.size() + " records");
    }

    private static String processData(String data) {
        // Simulate expensive processing
        return data.toUpperCase();
    }
}

Complete Java Example: ETL Pipeline

import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.*;
import java.util.concurrent.*;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

public class JSONLETLPipeline {
    private static final ObjectMapper mapper = new ObjectMapper();

    public static class InputRecord {
        public int id;
        public String status;
        public int value;
    }

    public static class OutputRecord {
        public int id;
        public boolean processed;
        public int result;

        public OutputRecord(int id, boolean processed, int result) {
            this.id = id;
            this.processed = processed;
            this.result = result;
        }
    }

    public static void main(String[] args) throws Exception {
        runETL("input.jsonl.gz", "output.jsonl.gz", 10);
    }

    public static void runETL(String inputPath, String outputPath, int numThreads)
            throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(numThreads);
        BlockingQueue<OutputRecord> outputQueue = new LinkedBlockingQueue<>(1000);

        // Writer thread
        CompletableFuture<Void> writerFuture = CompletableFuture.runAsync(() -> {
            try (BufferedWriter writer = new BufferedWriter(
                    new OutputStreamWriter(
                        new GZIPOutputStream(new FileOutputStream(outputPath))))) {

                while (true) {
                    OutputRecord record = outputQueue.poll(1, TimeUnit.SECONDS);
                    if (record == null) continue;

                    if (record.id == -1) break; // Sentinel value

                    String json = mapper.writeValueAsString(record);
                    writer.write(json);
                    writer.newLine();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });

        // Read and process
        try (BufferedReader reader = new BufferedReader(
                new InputStreamReader(
                    new GZIPInputStream(new FileInputStream(inputPath))))) {

            reader.lines()
                .parallel()
                .filter(line -> !line.trim().isEmpty())
                .forEach(line -> {
                    try {
                        InputRecord input = mapper.readValue(line, InputRecord.class);

                        // Filter
                        if (!"active".equals(input.status)) {
                            return;
                        }

                        // Transform
                        OutputRecord output = new OutputRecord(
                            input.id,
                            true,
                            input.value * 2
                        );

                        outputQueue.put(output);
                    } catch (Exception e) {
                        System.err.println("Error: " + e.getMessage());
                    }
                });
        }

        // Signal completion
        outputQueue.put(new OutputRecord(-1, false, 0));

        // Wait for writer
        writerFuture.get();
        executor.shutdown();

        System.out.println("ETL pipeline complete");
    }
}

Best Practices

Use BufferedReader/Writer: Provides significant performance improvement for file I/O
Reuse ObjectMapper: Creating new instances is expensive - use a singleton
Handle errors gracefully: One bad line shouldn't crash your entire pipeline
Use try-with-resources: Ensures proper cleanup of file handles
Consider streaming for large files: Process line-by-line to minimize memory usage

Rust

Setup and Dependencies

Add serde and serde_json to your Cargo.toml:

[dependencies]
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"

Reading JSONL

use serde::{Deserialize, Serialize};
use std::fs::File;
use std::io::{BufRead, BufReader};

#[derive(Debug, Deserialize, Serialize)]
struct User {
    id: u32,
    name: String,
    email: String,
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let file = File::open("data.jsonl")?;
    let reader = BufReader::new(file);

    for (line_num, line) in reader.lines().enumerate() {
        let line = line?;

        // Skip empty lines
        if line.trim().is_empty() {
            continue;
        }

        // Parse JSON
        match serde_json::from_str::<User>(&line) {
            Ok(user) => {
                println!("User: {} ({})", user.name, user.email);
            }
            Err(e) => {
                eprintln!("Error on line {}: {}", line_num + 1, e);
            }
        }
    }

    Ok(())
}

Writing JSONL

use serde::{Deserialize, Serialize};
use std::fs::File;
use std::io::{BufWriter, Write};

#[derive(Debug, Deserialize, Serialize)]
struct User {
    id: u32,
    name: String,
    email: String,
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let users = vec![
        User {
            id: 1,
            name: "Alice".to_string(),
            email: "[email protected]".to_string(),
        },
        User {
            id: 2,
            name: "Bob".to_string(),
            email: "[email protected]".to_string(),
        },
    ];

    let file = File::create("output.jsonl")?;
    let mut writer = BufWriter::new(file);

    for user in users {
        // Serialize to JSON string
        let json = serde_json::to_string(&user)?;

        // Write JSON + newline
        writeln!(writer, "{}", json)?;
    }

    // Ensure all data is written
    writer.flush()?;

    println!("Successfully wrote JSONL file");
    Ok(())
}

Memory-Efficient Streaming

Process large files line-by-line with minimal memory:

use serde::{Deserialize, Serialize};
use std::fs::File;
use std::io::{BufRead, BufReader};

#[derive(Debug, Deserialize)]
struct Record {
    id: u32,
    data: String,
}

fn stream_jsonl<F>(filepath: &str, mut processor: F) -> Result<(), Box<dyn std::error::Error>>
where
    F: FnMut(Record) -> Result<(), Box<dyn std::error::Error>>,
{
    let file = File::open(filepath)?;
    let reader = BufReader::new(file);

    for line in reader.lines() {
        let line = line?;

        if line.trim().is_empty() {
            continue;
        }

        match serde_json::from_str::<Record>(&line) {
            Ok(record) => {
                processor(record)?;
            }
            Err(e) => {
                eprintln!("Parse error: {}", e);
                continue;
            }
        }
    }

    Ok(())
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Process each record as it's read
    stream_jsonl("large.jsonl", |record| {
        println!("Processing ID: {}", record.id);
        // Do expensive work here
        Ok(())
    })?;

    Ok(())
}

Robust Error Handling

use serde::{Deserialize, Serialize};
use std::fs::File;
use std::io::{BufRead, BufReader};

#[derive(Debug, Deserialize)]
struct User {
    id: u32,
    name: String,
}

#[derive(Debug)]
struct ParseStats {
    total_lines: usize,
    successful: usize,
    errors: usize,
}

fn read_jsonl_with_stats(filepath: &str) -> Result<(Vec<User>, ParseStats), std::io::Error> {
    let file = File::open(filepath)?;
    let reader = BufReader::new(file);

    let mut users = Vec::new();
    let mut stats = ParseStats {
        total_lines: 0,
        successful: 0,
        errors: 0,
    };

    for (line_num, line) in reader.lines().enumerate() {
        stats.total_lines += 1;
        let line = line?;

        if line.trim().is_empty() {
            continue;
        }

        match serde_json::from_str::<User>(&line) {
            Ok(user) => {
                users.push(user);
                stats.successful += 1;
            }
            Err(e) => {
                eprintln!("Line {}: {}", line_num + 1, e);
                stats.errors += 1;
            }
        }
    }

    Ok((users, stats))
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (users, stats) = read_jsonl_with_stats("data.jsonl")?;

    println!("Loaded {} users", users.len());
    println!("Stats: {:?}", stats);

    Ok(())
}

Compressed JSONL

Add flate2 for gzip support:

// Cargo.toml
[dependencies]
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
flate2 = "1.0"

// src/main.rs
use flate2::read::GzDecoder;
use flate2::write::GzEncoder;
use flate2::Compression;
use serde::{Deserialize, Serialize};
use std::fs::File;
use std::io::{BufRead, BufReader, BufWriter, Write};

#[derive(Debug, Deserialize, Serialize)]
struct Record {
    id: u32,
    data: String,
}

fn read_gzipped_jsonl(filepath: &str) -> Result<Vec<Record>, Box<dyn std::error::Error>> {
    let file = File::open(filepath)?;
    let decoder = GzDecoder::new(file);
    let reader = BufReader::new(decoder);

    let mut records = Vec::new();

    for line in reader.lines() {
        let line = line?;
        if line.trim().is_empty() {
            continue;
        }

        let record: Record = serde_json::from_str(&line)?;
        records.push(record);
    }

    Ok(records)
}

fn write_gzipped_jsonl(records: &[Record], filepath: &str) -> Result<(), Box<dyn std::error::Error>> {
    let file = File::create(filepath)?;
    let encoder = GzEncoder::new(file, Compression::default());
    let mut writer = BufWriter::new(encoder);

    for record in records {
        let json = serde_json::to_string(record)?;
        writeln!(writer, "{}", json)?;
    }

    writer.flush()?;
    Ok(())
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Read gzipped JSONL
    let records = read_gzipped_jsonl("data.jsonl.gz")?;
    println!("Loaded {} records", records.len());

    // Write gzipped JSONL
    write_gzipped_jsonl(&records, "output.jsonl.gz")?;
    println!("Wrote compressed JSONL");

    Ok(())
}

Parallel Processing with Rayon

Add rayon for easy parallelism:

// Cargo.toml
[dependencies]
rayon = "1.7"
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"

// src/main.rs
use rayon::prelude::*;
use serde::{Deserialize, Serialize};
use std::fs::File;
use std::io::{BufRead, BufReader};

#[derive(Debug, Deserialize, Serialize)]
struct Record {
    id: u32,
    status: String,
    value: i32,
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let file = File::open("data.jsonl")?;
    let reader = BufReader::new(file);

    // Read all lines into memory
    let lines: Vec<String> = reader.lines().filter_map(|l| l.ok()).collect();

    // Process in parallel
    let results: Vec<Record> = lines
        .par_iter()
        .filter_map(|line| {
            if line.trim().is_empty() {
                return None;
            }

            serde_json::from_str::<Record>(line).ok()
        })
        .filter(|record| record.status == "active")
        .map(|mut record| {
            // Expensive processing
            record.value *= 2;
            record
        })
        .collect();

    println!("Processed {} records", results.len());

    Ok(())
}

Complete Rust Example: ETL Pipeline

use flate2::read::GzDecoder;
use flate2::write::GzEncoder;
use flate2::Compression;
use rayon::prelude::*;
use serde::{Deserialize, Serialize};
use std::fs::File;
use std::io::{BufRead, BufReader, BufWriter, Write};

#[derive(Debug, Deserialize)]
struct InputRecord {
    id: u32,
    status: String,
    value: i32,
}

#[derive(Debug, Serialize)]
struct OutputRecord {
    id: u32,
    processed: bool,
    result: i32,
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    run_etl("input.jsonl.gz", "output.jsonl.gz")?;
    Ok(())
}

fn run_etl(input_path: &str, output_path: &str) -> Result<(), Box<dyn std::error::Error>> {
    // Read gzipped input
    let input_file = File::open(input_path)?;
    let decoder = GzDecoder::new(input_file);
    let reader = BufReader::new(decoder);

    // Read all lines
    let lines: Vec<String> = reader.lines().filter_map(|l| l.ok()).collect();

    println!("Processing {} lines...", lines.len());

    // Parallel ETL pipeline
    let results: Vec<OutputRecord> = lines
        .par_iter()
        .filter_map(|line| {
            if line.trim().is_empty() {
                return None;
            }

            // Parse
            serde_json::from_str::<InputRecord>(line).ok()
        })
        .filter(|input| {
            // Filter: only active records
            input.status == "active"
        })
        .map(|input| {
            // Transform
            OutputRecord {
                id: input.id,
                processed: true,
                result: input.value * 2,
            }
        })
        .collect();

    println!("Filtered to {} records", results.len());

    // Write gzipped output
    let output_file = File::create(output_path)?;
    let encoder = GzEncoder::new(output_file, Compression::default());
    let mut writer = BufWriter::new(encoder);

    for record in results {
        let json = serde_json::to_string(&record)?;
        writeln!(writer, "{}", json)?;
    }

    writer.flush()?;
    println!("ETL complete!");

    Ok(())
}

Best Practices

Use BufReader/BufWriter: Dramatically improves I/O performance
Leverage serde's derive macro: Auto-generate serialization code
Handle errors explicitly: Use Result types and proper error propagation
Use Rayon for parallelism: Easy parallel processing with minimal code changes
Flush writers explicitly: Ensures all data is written to disk
Consider memory vs speed tradeoffs: Line-by-line for huge files, bulk for better parallelism

C#

Setup and Dependencies

System.Text.Json ships in the box on modern .NET, so a JSONL reader needs no package reference at all:

using System.Text.Json;
using System.Text.Json.Serialization;

// System.Text.Json is part of the shared framework on .NET Core 3.0
// and every later release. Nothing to install.
//
// Only needed when you target .NET Framework or .NET Standard 2.0:
//     dotnet add package System.Text.Json
//
// Legacy option, still very widely deployed:
//     dotnet add package Newtonsoft.Json
// Newtonsoft.Json (Json.NET) works fine for JSONL and its JsonTextReader
// has a SupportMultipleContent flag built for exactly this format. It is
// slower and allocates more than System.Text.Json, so treat it as the
// choice for existing code, not for new code.

// Cache one options instance. Constructing JsonSerializerOptions per
// call throws away the reflection metadata cache and is the single
// most common System.Text.Json performance mistake.
public static class Json
{
    public static readonly JsonSerializerOptions Options = new()
    {
        PropertyNameCaseInsensitive = true,
        DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull
    };
}

public sealed record User(int Id, string Name, string Email);

Reading JSONL

File.ReadLinesAsync hands you one line at a time and never materializes the file, which is the whole point of JSONL:

using System.Text.Json;

// File.ReadLinesAsync returns IAsyncEnumerable<string> and reads lazily.
// Available on .NET 7 and later.
await foreach (string line in File.ReadLinesAsync("data.jsonl"))
{
    if (string.IsNullOrWhiteSpace(line))
    {
        continue;
    }

    User? user = JsonSerializer.Deserialize<User>(line, Json.Options);
    Console.WriteLine(user?.Name);
}

// Synchronous equivalent. File.ReadLines is lazy too - do not confuse it
// with File.ReadAllLines, which loads the entire file into a string[].
foreach (string line in File.ReadLines("data.jsonl"))
{
    if (string.IsNullOrWhiteSpace(line))
    {
        continue;
    }

    User? user = JsonSerializer.Deserialize<User>(line, Json.Options);
    Console.WriteLine(user?.Name);
}

// Materialize into a list only when the file genuinely fits in memory
List<User> users = new();

await foreach (string line in File.ReadLinesAsync("data.jsonl"))
{
    if (string.IsNullOrWhiteSpace(line))
    {
        continue;
    }

    if (JsonSerializer.Deserialize<User>(line, Json.Options) is { } user)
    {
        users.Add(user);
    }
}

// Schemaless records: deserialize to JsonDocument or a dictionary
using JsonDocument doc = JsonDocument.Parse(line);
string name = doc.RootElement.GetProperty("name").GetString() ?? "";

Dictionary<string, object?>? loose =
    JsonSerializer.Deserialize<Dictionary<string, object?>>(line, Json.Options);

Streaming with DeserializeAsyncEnumerable

The topLevelValues overload reads a stream of whitespace-delimited root-level JSON values, which is precisely the JSONL shape. Nothing splits lines by hand and nothing buffers the file. It needs .NET 9 or newer; on older runtimes use the ReadLinesAsync loop above.

using System.Runtime.CompilerServices;
using System.Text.Json;

await using FileStream stream = File.OpenRead("huge.jsonl");

IAsyncEnumerable<User?> source = JsonSerializer.DeserializeAsyncEnumerable<User>(
    stream,
    topLevelValues: true,
    Json.Options);

await foreach (User? user in source)
{
    if (user is null)
    {
        continue;
    }

    await ProcessAsync(user);
}

// Batched streaming for bulk inserts, with cancellation plumbed through
static async IAsyncEnumerable<List<User>> ReadBatchesAsync(
    string path,
    int batchSize,
    [EnumeratorCancellation] CancellationToken token = default)
{
    await using FileStream stream = File.OpenRead(path);

    IAsyncEnumerable<User?> source = JsonSerializer.DeserializeAsyncEnumerable<User>(
        stream, topLevelValues: true, Json.Options, token);

    List<User> batch = new(batchSize);

    await foreach (User? user in source.WithCancellation(token))
    {
        if (user is null)
        {
            continue;
        }

        batch.Add(user);

        if (batch.Count >= batchSize)
        {
            yield return batch;
            batch = new List<User>(batchSize);
        }
    }

    if (batch.Count > 0)
    {
        yield return batch;
    }
}

// Usage
await foreach (List<User> batch in ReadBatchesAsync("huge.jsonl", 5_000))
{
    await BulkInsertAsync(batch);
}

// The same call works on an HTTP response, so a streaming JSONL API is
// consumed record by record instead of buffered end to end.
using HttpResponseMessage response = await http.GetAsync(
    "https://api.example.com/data.jsonl",
    HttpCompletionOption.ResponseHeadersRead);

await using Stream body = await response.Content.ReadAsStreamAsync();

await foreach (User? user in JsonSerializer.DeserializeAsyncEnumerable<User>(
    body, topLevelValues: true, Json.Options))
{
    Console.WriteLine(user?.Name);
}

Note: without topLevelValues: true the same method expects a single JSON array and will fail on the second line of a JSONL file.

Writing JSONL

using System.Text.Json;

// JsonSerializer.Serialize emits no newline unless WriteIndented is on,
// so one Serialize call per line is always a valid JSONL record.
await using StreamWriter writer = new("output.jsonl");

foreach (User user in users)
{
    await writer.WriteLineAsync(JsonSerializer.Serialize(user, Json.Options));
}

// Append to an existing file
await using StreamWriter appender = new("output.jsonl", append: true);
await appender.WriteLineAsync(JsonSerializer.Serialize(newUser, Json.Options));

// GOTCHA: WriteLineAsync uses Environment.NewLine, which is CRLF on
// Windows. Most JSONL consumers tolerate it, but pin LF when the file
// crosses platforms or gets hashed for comparison.
await using StreamWriter lfWriter = new("output.jsonl")
{
    NewLine = "\n"
};

foreach (User user in users)
{
    await lfWriter.WriteLineAsync(JsonSerializer.Serialize(user, Json.Options));
}

// Lowest-allocation path: serialize straight to UTF-8 bytes and skip
// the string encode/decode round trip entirely.
await using FileStream raw = File.Create("output.jsonl");

foreach (User user in users)
{
    byte[] utf8 = JsonSerializer.SerializeToUtf8Bytes(user, Json.Options);
    await raw.WriteAsync(utf8);
    raw.WriteByte((byte)'\n');
}

// Gzipped output
using System.IO.Compression;

await using FileStream outFile = File.Create("output.jsonl.gz");
await using GZipStream gzip = new(outFile, CompressionLevel.Optimal);
await using StreamWriter gzWriter = new(gzip) { NewLine = "\n" };

foreach (User user in users)
{
    await gzWriter.WriteLineAsync(JsonSerializer.Serialize(user, Json.Options));
}

Robust Error Handling

One malformed line should cost you one record, not the whole run. Catch JsonException per line and keep going:

using System.Text.Json;

public sealed record ParseStats(int Total, int Parsed, int Failed);

static async Task<(List<User> Users, ParseStats Stats)> ReadWithStatsAsync(string path)
{
    List<User> users = new();
    int total = 0, parsed = 0, failed = 0;
    int lineNumber = 0;

    await foreach (string line in File.ReadLinesAsync(path))
    {
        lineNumber++;

        if (string.IsNullOrWhiteSpace(line))
        {
            continue;
        }

        total++;

        try
        {
            User? user = JsonSerializer.Deserialize<User>(line, Json.Options);

            if (user is null)
            {
                // A literal "null" line parses successfully and yields null.
                // Treat it as bad data rather than letting it flow downstream.
                Console.Error.WriteLine($"Line {lineNumber}: null record");
                failed++;
                continue;
            }

            users.Add(user);
            parsed++;
        }
        catch (JsonException ex)
        {
            // ex.LineNumber and ex.BytePositionInLine are relative to the
            // fragment handed to the serializer, not to the file, so log
            // the file line number yourself.
            Console.Error.WriteLine($"Line {lineNumber}: {ex.Message}");
            failed++;
        }
    }

    return (users, new ParseStats(total, parsed, failed));
}

// Usage
(List<User> users, ParseStats stats) = await ReadWithStatsAsync("data.jsonl");
Console.WriteLine($"Parsed {stats.Parsed} of {stats.Total}, {stats.Failed} bad lines");

// Fail fast instead, when a single bad line means the upstream job broke
static async Task<List<User>> ReadStrictAsync(string path)
{
    List<User> users = new();
    int lineNumber = 0;

    await foreach (string line in File.ReadLinesAsync(path))
    {
        lineNumber++;

        if (string.IsNullOrWhiteSpace(line))
        {
            continue;
        }

        try
        {
            users.Add(JsonSerializer.Deserialize<User>(line, Json.Options)!);
        }
        catch (JsonException ex)
        {
            throw new InvalidDataException($"Malformed JSONL at line {lineNumber}", ex);
        }
    }

    return users;
}

Complete C# Example: Streaming ETL Pipeline

using System.IO.Compression;
using System.Text.Json;
using System.Text.Json.Serialization;

// Reads gzipped JSONL, filters, transforms, writes gzipped JSONL.
// Memory stays flat regardless of input size.
//
//     dotnet run -- input.jsonl.gz output.jsonl.gz

public sealed record InputRecord(int Id, string Status, int Value);
public sealed record OutputRecord(int Id, bool Processed, int Result);

public static class JsonlEtl
{
    private static readonly JsonSerializerOptions Options = new()
    {
        PropertyNameCaseInsensitive = true,
        DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull
    };

    public static async Task Main(string[] args)
    {
        string input = args.Length > 0 ? args[0] : "input.jsonl.gz";
        string output = args.Length > 1 ? args[1] : "output.jsonl.gz";

        await RunAsync(input, output);
    }

    public static async Task RunAsync(string inputPath, string outputPath)
    {
        await using FileStream inFile = File.OpenRead(inputPath);
        await using GZipStream gunzip = new(inFile, CompressionMode.Decompress);

        await using FileStream outFile = File.Create(outputPath);
        await using GZipStream gzip = new(outFile, CompressionLevel.Optimal);
        await using StreamWriter writer = new(gzip) { NewLine = "\n" };

        int read = 0;
        int written = 0;
        int skipped = 0;

        IAsyncEnumerable<InputRecord?> source =
            JsonSerializer.DeserializeAsyncEnumerable<InputRecord>(
                gunzip, topLevelValues: true, Options);

        await foreach (InputRecord? input in source)
        {
            read++;

            // Filter
            if (input is null || !string.Equals(input.Status, "active", StringComparison.Ordinal))
            {
                skipped++;
                continue;
            }

            // Transform
            OutputRecord output = new(input.Id, true, input.Value * 2);

            // Load
            await writer.WriteLineAsync(JsonSerializer.Serialize(output, Options));
            written++;
        }

        await writer.FlushAsync();

        Console.WriteLine($"Read {read}, wrote {written}, skipped {skipped}");
    }
}

Best Practices

Cache JsonSerializerOptions: A new instance per call discards the metadata cache and costs an order of magnitude in throughput
Prefer ReadLines over ReadAllLines: One is lazy, the other loads the whole file, and the names are one character apart
Set StreamWriter.NewLine: The default is CRLF on Windows, which surprises consumers that split on LF
Use source generation for hot paths: A JsonSerializerContext removes reflection and works under trimming and AOT
Catch JsonException per line: A single bad record should not abort a million-line file

Ruby

Setup and Dependencies

Ruby needs no dependency for JSONL at all. The honest recommendation is the standard library JSON module driven by IO#each_line:

# JSON is in the Ruby standard library. Nothing to install.
require 'json'

# Optional accelerator, roughly 2-4x faster on parse-heavy work:
#   gem install oj
# Gemfile:
#   gem 'oj'

# Deliberately NOT recommended: the ndjson gem. Its repository has had
# no pushes since 2022, so it is unmaintained, and it wraps three lines
# of stdlib code. Reach for a gem when it buys you something; here it
# only buys you an abandoned dependency.
#
# The whole "library" it replaces:
#
#   File.foreach(path) { |line| JSON.parse(line) }

Reading JSONL

IO#each_line streams the handle one line at a time, so memory stays flat no matter how big the file is. File.foreach is the same iterator plus automatic close:

require 'json'

# Simplest correct reader
File.foreach('data.jsonl', encoding: 'UTF-8') do |line|
  line = line.strip
  next if line.empty?

  record = JSON.parse(line)
  puts record['name']
end

# Symbol keys instead of string keys
File.foreach('data.jsonl', encoding: 'UTF-8') do |line|
  next if line.strip.empty?

  record = JSON.parse(line, symbolize_names: true)
  puts record[:name]
end

# Explicit handle when you also want the line number
File.open('data.jsonl', 'r:UTF-8') do |file|
  file.each_line.with_index(1) do |line, line_number|
    next if line.strip.empty?

    record = JSON.parse(line)
    puts "#{line_number}: #{record['name']}"
  end
end

# Gzipped input reads the same way - GzipReader is an IO
require 'zlib'

Zlib::GzipReader.open('data.jsonl.gz') do |gz|
  gz.each_line do |line|
    next if line.strip.empty?

    record = JSON.parse(line)
    puts record['name']
  end
end

# NEVER do this on a real file - it loads everything at once
# records = File.readlines('data.jsonl').map { |l| JSON.parse(l) }

Lazy Enumerators for Huge Files

Return an Enumerator when no block is given and the reader composes with Ruby's lazy pipeline operators:

require 'json'

def stream_jsonl(path)
  return enum_for(:stream_jsonl, path) unless block_given?

  File.foreach(path, encoding: 'UTF-8') do |line|
    line = line.strip
    next if line.empty?

    yield JSON.parse(line)
  end
end

# Lazy pipeline over a file larger than RAM. Only 100 records are ever
# parsed past the first match because .lazy stops pulling.
active_names = stream_jsonl('huge.jsonl')
               .lazy
               .select { |record| record['status'] == 'active' }
               .map    { |record| record['name'] }
               .first(100)

# Batch for bulk database inserts
stream_jsonl('huge.jsonl').each_slice(5_000) do |batch|
  User.insert_all(batch)
end

# Filter and rewrite without ever holding two records at once
def filter_jsonl(input_path, output_path)
  File.open(output_path, 'w:UTF-8') do |out|
    stream_jsonl(input_path) do |record|
      next unless yield(record)

      out.puts(JSON.generate(record))
    end
  end
end

filter_jsonl('users.jsonl', 'active_users.jsonl') do |record|
  record['status'] == 'active'
end

Writing JSONL

require 'json'

records = [
  { 'id' => 1, 'name' => 'Alice', 'age' => 30 },
  { 'id' => 2, 'name' => 'Bob',   'age' => 25 }
]

# IO#puts appends a newline only when the string does not already end
# in one, so it can never emit a blank JSONL record by accident.
File.open('output.jsonl', 'w:UTF-8') do |file|
  records.each { |record| file.puts(JSON.generate(record)) }
end

# Append to an existing file
File.open('output.jsonl', 'a:UTF-8') do |file|
  file.puts(JSON.generate({ 'id' => 3, 'name' => 'Charlie' }))
end

# JSON.generate keeps UTF-8 intact by default. Pass ascii_only: true
# only when the consumer really requires escaped \uXXXX output.
File.open('output.jsonl', 'w:UTF-8') do |file|
  records.each { |record| file.puts(JSON.generate(record, ascii_only: true)) }
end

# Gzipped output
require 'zlib'

Zlib::GzipWriter.open('output.jsonl.gz') do |gz|
  records.each { |record| gz.puts(JSON.generate(record)) }
end

# Streaming an endless source to JSONL - flush so a crash does not
# lose the tail of the buffer.
File.open('events.jsonl', 'a:UTF-8') do |file|
  event_source.each do |event|
    file.puts(JSON.generate(event))
    file.flush
  end
end

Robust Error Handling

JSON.parse raises JSON::ParserError. Rescue it per line so one bad record cannot end the run:

require 'json'

Stats = Struct.new(:total, :parsed, :failed)

def read_jsonl_with_stats(path)
  records = []
  stats = Stats.new(0, 0, 0)

  File.foreach(path, encoding: 'UTF-8').with_index(1) do |line, line_number|
    line = line.strip
    next if line.empty?

    stats.total += 1

    begin
      records << JSON.parse(line)
      stats.parsed += 1
    rescue JSON::ParserError => e
      warn "Line #{line_number}: #{e.message}"
      stats.failed += 1
    end
  end

  [records, stats]
end

records, stats = read_jsonl_with_stats('data.jsonl')
puts "parsed #{stats.parsed} of #{stats.total} (#{stats.failed} bad lines)"

# Route bad lines to a quarantine file instead of dropping them
def read_jsonl_quarantined(path, reject_path)
  File.open(reject_path, 'w:UTF-8') do |rejects|
    File.foreach(path, encoding: 'UTF-8') do |line|
      line = line.strip
      next if line.empty?

      begin
        yield JSON.parse(line)
      rescue JSON::ParserError
        rejects.puts(line)
      end
    end
  end
end

# Invalid UTF-8 raises before JSON ever sees the line. Scrub it first
# when the upstream producer is not trustworthy.
File.foreach('messy.jsonl', encoding: 'UTF-8') do |line|
  line = line.scrub.strip
  next if line.empty?

  record = JSON.parse(line)
end

High-Performance Parsing (oj)

oj is the fast, actively maintained JSON library for Ruby. It is worth adding when parsing dominates your profile, and it is a drop-in change:

# gem install oj
require 'oj'

# :strict mode maps JSON to plain Hash and Array with no Ruby object
# instantiation, which is what you want for untrusted input.
File.foreach('data.jsonl', encoding: 'UTF-8') do |line|
  next if line.strip.empty?

  record = Oj.load(line, mode: :strict)
  puts record['name']
end

# Writing
File.open('output.jsonl', 'w:UTF-8') do |file|
  records.each { |record| file.puts(Oj.dump(record, mode: :strict)) }
end

# Compatibility shim: makes JSON.parse and JSON.generate use Oj
# internally, so existing code gets faster without being edited.
require 'oj'
Oj.mimic_JSON

# Oj raises Oj::ParseError. Rescue both when a code path might run
# against either backend.
begin
  record = Oj.load(line, mode: :strict)
rescue Oj::ParseError, JSON::ParserError => e
  warn "bad line: #{e.message}"
end

Reach for it when: you have profiled and JSON parsing is the bottleneck. Stdlib JSON is fast enough for most JSONL workloads, and one fewer native extension is one fewer thing to rebuild.

Complete Ruby Example: Streaming ETL Pipeline

#!/usr/bin/env ruby
# frozen_string_literal: true

# Reads gzipped JSONL, filters, transforms, writes gzipped JSONL.
# Constant memory: one record is live at a time.
#
#   ruby etl.rb input.jsonl.gz output.jsonl.gz

require 'json'
require 'zlib'

class JsonlEtl
  Stats = Struct.new(:read, :written, :failed)

  def initialize(input_path, output_path)
    @input_path = input_path
    @output_path = output_path
    @stats = Stats.new(0, 0, 0)
  end

  def run
    Zlib::GzipWriter.open(@output_path) do |out|
      each_record do |record|
        next unless keep?(record)

        out.puts(JSON.generate(transform(record)))
        @stats.written += 1
      end
    end

    @stats
  end

  private

  def each_record
    Zlib::GzipReader.open(@input_path) do |gz|
      gz.each_line.with_index(1) do |line, line_number|
        line = line.strip
        next if line.empty?

        @stats.read += 1

        begin
          yield JSON.parse(line)
        rescue JSON::ParserError => e
          warn "Skipping line #{line_number}: #{e.message}"
          @stats.failed += 1
        end
      end
    end
  end

  def keep?(record)
    record['status'] == 'active'
  end

  def transform(record)
    {
      'id' => record['id'],
      'processed' => true,
      'result' => record['value'].to_i * 2
    }
  end
end

if __FILE__ == $PROGRAM_NAME
  input = ARGV[0] || 'input.jsonl.gz'
  output = ARGV[1] || 'output.jsonl.gz'

  stats = JsonlEtl.new(input, output).run
  puts "read #{stats.read}, wrote #{stats.written}, failed #{stats.failed}"
end

Best Practices

Use File.foreach, never File.readlines: One streams, the other builds an array of every line in the file
Write with IO#puts: It adds a newline only when one is missing, so it cannot emit an empty record
Return an Enumerator: The enum_for guard lets callers chain lazy, each_slice, and with_index for free
Set the encoding explicitly: Open with 'r:UTF-8' rather than relying on the ambient default external encoding
Skip the ndjson gem: It is unmaintained and replaces code you can read in one line

PHP

Setup and Dependencies

json_decode and json_encode are part of the PHP core, so a JSONL reader is dependency free. Only very large single lines justify a library:

<?php
declare(strict_types=1);

// ext-json is compiled into PHP and cannot be disabled as of PHP 8.0.
// Nothing to install for ordinary JSONL work.

// Only when a single LINE is itself an enormous document:
//     composer require halaxa/json-machine

// Always pass JSON_THROW_ON_ERROR. Without it json_decode returns null
// on failure, which is indistinguishable from a valid "null" record and
// forces you to call json_last_error() after every single decode.
const JSONL_DECODE_FLAGS = JSON_THROW_ON_ERROR;

const JSONL_ENCODE_FLAGS = JSON_THROW_ON_ERROR
    | JSON_UNESCAPED_SLASHES
    | JSON_UNESCAPED_UNICODE;

Reading JSONL with Generators

A generator plus fgets is the idiomatic PHP streaming reader. Memory stays flat because only one line is ever resident:

<?php
declare(strict_types=1);

/**
 * Stream a JSONL file, yielding line number => decoded record.
 */
function readJsonl(string $path): Generator
{
    $handle = fopen($path, 'rb');

    if ($handle === false) {
        throw new RuntimeException("Cannot open {$path}");
    }

    try {
        $lineNumber = 0;

        while (($line = fgets($handle)) !== false) {
            $lineNumber++;
            $line = trim($line);

            if ($line === '') {
                continue;
            }

            yield $lineNumber => json_decode($line, true, 512, JSON_THROW_ON_ERROR);
        }
    } finally {
        // finally runs even when the consumer breaks out of the foreach
        fclose($handle);
    }
}

foreach (readJsonl('data.jsonl') as $lineNumber => $record) {
    echo $record['name'], PHP_EOL;
}

// Bound memory against a pathological single line by capping fgets
while (($line = fgets($handle, 1048576)) !== false) {
    // ...
}

// Gzipped input: gzopen and gzgets are drop-in replacements
$handle = gzopen('data.jsonl.gz', 'rb');

while (($line = gzgets($handle)) !== false) {
    $line = trim($line);

    if ($line !== '') {
        $record = json_decode($line, true, 512, JSON_THROW_ON_ERROR);
    }
}

gzclose($handle);

// NEVER do this on a real file - file() builds an array of every line
// $lines = file('data.jsonl', FILE_IGNORE_NEW_LINES);

Writing JSONL

<?php
declare(strict_types=1);

/**
 * Write any iterable of records as JSONL. Accepts a generator, so the
 * producer never has to build the full array either.
 */
function writeJsonl(iterable $records, string $path, string $mode = 'wb'): int
{
    $handle = fopen($path, $mode);

    if ($handle === false) {
        throw new RuntimeException("Cannot write {$path}");
    }

    $written = 0;

    try {
        foreach ($records as $record) {
            // json_encode never emits a raw newline, so one call per
            // line always produces exactly one valid JSONL record.
            $json = json_encode(
                $record,
                JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE
            );

            fwrite($handle, $json . "\n");
            $written++;
        }
    } finally {
        fclose($handle);
    }

    return $written;
}

writeJsonl([
    ['id' => 1, 'name' => 'Alice'],
    ['id' => 2, 'name' => 'Bob'],
], 'output.jsonl');

// Append instead of overwrite
writeJsonl([['id' => 3, 'name' => 'Charlie']], 'output.jsonl', 'ab');

// Gzipped output
$out = gzopen('output.jsonl.gz', 'wb9');

foreach ($records as $record) {
    gzwrite($out, json_encode($record, JSON_THROW_ON_ERROR) . "\n");
}

gzclose($out);

// Streaming a JSONL HTTP response. Flush per record so the client
// receives rows as they are produced instead of at the end.
header('Content-Type: application/x-ndjson');

foreach (readJsonl('data.jsonl') as $record) {
    echo json_encode($record, JSON_THROW_ON_ERROR), "\n";
    flush();
}

Robust Error Handling

With JSON_THROW_ON_ERROR a bad line raises JsonException. Catch it per line and keep the run alive:

<?php
declare(strict_types=1);

final class JsonlStats
{
    public function __construct(
        public int $total = 0,
        public int $parsed = 0,
        public int $failed = 0,
    ) {
    }
}

function readJsonlTolerant(string $path, JsonlStats $stats): Generator
{
    $handle = fopen($path, 'rb');

    if ($handle === false) {
        throw new RuntimeException("Cannot open {$path}");
    }

    try {
        $lineNumber = 0;

        while (($line = fgets($handle)) !== false) {
            $lineNumber++;
            $line = trim($line);

            if ($line === '') {
                continue;
            }

            $stats->total++;

            try {
                $record = json_decode($line, true, 512, JSON_THROW_ON_ERROR);
            } catch (JsonException $e) {
                fwrite(STDERR, "Line {$lineNumber}: {$e->getMessage()}\n");
                $stats->failed++;
                continue;
            }

            $stats->parsed++;

            yield $record;
        }
    } finally {
        fclose($handle);
    }
}

$stats = new JsonlStats();

foreach (readJsonlTolerant('data.jsonl', $stats) as $record) {
    // process one record
}

printf(
    "parsed %d of %d, %d bad lines%s",
    $stats->parsed,
    $stats->total,
    $stats->failed,
    PHP_EOL
);

// Encoding fails too: INF, NAN, and invalid UTF-8 all throw on encode.
try {
    $json = json_encode($record, JSON_THROW_ON_ERROR);
} catch (JsonException $e) {
    // JSON_INVALID_UTF8_SUBSTITUTE replaces bad bytes instead of throwing
    $json = json_encode($record, JSON_THROW_ON_ERROR | JSON_INVALID_UTF8_SUBSTITUTE);
}

Very Large Lines with JsonMachine

JsonMachine parses one JSON document incrementally rather than building the whole PHP structure. It has no NDJSON mode, so pair it with the line reader above: fgets gives you the line, JsonMachine walks the array inside it.

<?php
// composer require halaxa/json-machine
declare(strict_types=1);

use JsonMachine\Items;
use JsonMachine\JsonDecoder\ExtJsonDecoder;

/** Yield raw, non-empty lines - no decoding. */
function readLines(string $path): Generator
{
    $handle = fopen($path, 'rb');

    if ($handle === false) {
        throw new RuntimeException("Cannot open {$path}");
    }

    try {
        while (($line = fgets($handle)) !== false) {
            $line = trim($line);

            if ($line !== '') {
                yield $line;
            }
        }
    } finally {
        fclose($handle);
    }
}

// Each line here is a session object holding a huge "events" array.
// json_decode would build every event as a PHP array at once;
// JsonMachine holds one at a time.
foreach (readLines('sessions.jsonl') as $line) {
    $events = Items::fromString($line, [
        'pointer' => '/events',
        'decoder' => new ExtJsonDecoder(true),
    ]);

    foreach ($events as $event) {
        handleEvent($event);
    }
}

// Items::fromStream works the same way when a single document arrives
// over the network rather than sitting on disk.
$stream = fopen('https://api.example.com/session.json', 'rb');

foreach (Items::fromStream($stream, ['pointer' => '/events']) as $event) {
    handleEvent($event);
}

fclose($stream);

Rule of thumb: if every line decodes comfortably on its own, plain json_decode is already the streaming solution. Add JsonMachine only when one line is too big for memory.

Complete PHP Example: Streaming ETL Pipeline

<?php
declare(strict_types=1);

/**
 * Reads gzipped JSONL, filters, transforms, writes gzipped JSONL.
 * Constant memory: exactly one record is live at a time.
 *
 *   php etl.php input.jsonl.gz output.jsonl.gz
 */

function readJsonlGz(string $path, callable $onError): Generator
{
    $handle = gzopen($path, 'rb');

    if ($handle === false) {
        throw new RuntimeException("Cannot open {$path}");
    }

    try {
        $lineNumber = 0;

        while (($line = gzgets($handle)) !== false) {
            $lineNumber++;
            $line = trim($line);

            if ($line === '') {
                continue;
            }

            try {
                $record = json_decode($line, true, 512, JSON_THROW_ON_ERROR);
            } catch (JsonException $e) {
                $onError($lineNumber, $e);
                continue;
            }

            yield $record;
        }
    } finally {
        gzclose($handle);
    }
}

function runEtl(string $inputPath, string $outputPath): void
{
    $out = gzopen($outputPath, 'wb9');

    if ($out === false) {
        throw new RuntimeException("Cannot write {$outputPath}");
    }

    $read = 0;
    $written = 0;
    $failed = 0;

    $onError = static function (int $lineNumber, JsonException $e) use (&$failed): void {
        fwrite(STDERR, "Skipping line {$lineNumber}: {$e->getMessage()}\n");
        $failed++;
    };

    try {
        foreach (readJsonlGz($inputPath, $onError) as $record) {
            $read++;

            // Filter
            if (($record['status'] ?? null) !== 'active') {
                continue;
            }

            // Transform
            $output = [
                'id' => $record['id'],
                'processed' => true,
                'result' => ((int) $record['value']) * 2,
            ];

            // Load
            gzwrite($out, json_encode($output, JSON_THROW_ON_ERROR) . "\n");
            $written++;
        }
    } finally {
        gzclose($out);
    }

    printf("read %d, wrote %d, failed %d%s", $read, $written, $failed, PHP_EOL);
}

runEtl($argv[1] ?? 'input.jsonl.gz', $argv[2] ?? 'output.jsonl.gz');

Best Practices

Always pass JSON_THROW_ON_ERROR: Otherwise a decode failure and a valid null record look identical
Return a Generator, not an array: The caller iterates the same way and memory stops scaling with file size
Close handles in a finally block: A consumer that breaks out of the foreach halts the generator mid-body
Avoid file() and file_get_contents(): Both load the entire file, which defeats the point of JSONL
Decode to arrays with the true flag: Associative arrays are faster and lighter than stdClass for bulk processing

JSONL by Language

Jump to a Language

Python

Reading JSONL (Standard Library)

Writing JSONL

High-Performance Parsing (orjson)

Working with Pandas

Streaming Large Files

Working with Compressed JSONL

Parallel Processing

Streaming with ijson

Using jsonlines Library

Complete Python Example: ETL Pipeline

JavaScript / Node.js

Reading JSONL (Node.js)

Writing JSONL

High-Performance Streaming (ndjson)

Transform Streams

Compressed JSONL

HTTP Streaming API

Browser Usage (Fetch API)

Complete Node.js Example: API + Streaming

Go

Reading JSONL (bufio.Scanner)

Writing JSONL

Streaming with json.Decoder

Concurrent Processing with Goroutines

Compressed JSONL

Generic JSONL Processing

Complete Go Example: Concurrent ETL Pipeline

Java

Setup and Dependencies

Reading JSONL with Jackson

Writing JSONL with Jackson

Streaming Large Files with Jackson

Reading JSONL with Gson

Writing JSONL with Gson

Parallel Processing with Streams

Complete Java Example: ETL Pipeline

Best Practices

Rust

Setup and Dependencies

Reading JSONL

Writing JSONL

Memory-Efficient Streaming

Robust Error Handling

Compressed JSONL

Parallel Processing with Rayon

Complete Rust Example: ETL Pipeline

Best Practices

C#

Setup and Dependencies

Reading JSONL

Streaming with DeserializeAsyncEnumerable

Writing JSONL

Robust Error Handling

Complete C# Example: Streaming ETL Pipeline

Best Practices

Ruby

Setup and Dependencies

Reading JSONL

Lazy Enumerators for Huge Files

Writing JSONL

Robust Error Handling

High-Performance Parsing (oj)

Complete Ruby Example: Streaming ETL Pipeline

Best Practices

PHP

Setup and Dependencies

Reading JSONL with Generators

Writing JSONL

Robust Error Handling

Very Large Lines with JsonMachine

Complete PHP Example: Streaming ETL Pipeline

Best Practices