Skip to main content

JSONL Troubleshooting Guide

Debug common errors, validate files, fix encoding issues, and optimize performance with comprehensive troubleshooting strategies

Common Issues & Solutions

Quick Diagnosis

Most JSONL problems fall into a few categories. This guide helps you quickly identify and fix issues with malformed JSON, encoding problems, parser errors, and performance bottlenecks. If you already have an exact error message in front of you, look it up in the JSONL error dictionary instead.

  • Validate JSONL file syntax
  • Detect encoding issues (UTF-8, BOM)
  • Fix malformed JSON records
  • Debug parser errors
  • Resolve memory and performance issues

Diagnostic Tools

Essential tools for troubleshooting JSONL files:

  • jq: Command-line JSON processor
  • jsonlint: JSON validation tool
  • grep/ripgrep: Search for patterns
  • hexdump: Inspect file encoding
  • py-spy: Profile Python performance

Jump to Topic

Validation

Encoding Issues

Malformed JSON

Parser Errors

Memory Issues

Performance

Data Quality

Tools & Scripts

File Validation

Python Validation Script 

import json
import sys
from typing import Dict, List, Tuple

class JSONLValidator:
    """Comprehensive JSONL file validator"""

    def __init__(self, filepath: str):
        self.filepath = filepath
        self.errors = []
        self.warnings = []
        self.stats = {
            'total_lines': 0,
            'valid_records': 0,
            'empty_lines': 0,
            'parse_errors': 0,
            'total_bytes': 0
        }

    def validate(self) -> Dict:
        """Validate entire JSONL file"""
        try:
            with open(self.filepath, 'r', encoding='utf-8') as f:
                line_num = 0

                for line in f:
                    line_num += 1
                    self.stats['total_lines'] += 1
                    self.stats['total_bytes'] += len(line.encode('utf-8'))

                    # Check for empty lines
                    if not line.strip():
                        self.stats['empty_lines'] += 1
                        self.warnings.append({
                            'line': line_num,
                            'type': 'empty_line',
                            'message': 'Empty line (should be removed)'
                        })
                        continue

                    # Check for trailing whitespace
                    if line != line.strip() + '\n' and line != line.strip():
                        self.warnings.append({
                            'line': line_num,
                            'type': 'whitespace',
                            'message': 'Line has leading or trailing whitespace'
                        })

                    # Validate JSON
                    try:
                        record = json.loads(line)
                        self.stats['valid_records'] += 1

                        # Check if record is object (not array or primitive)
                        if not isinstance(record, dict):
                            self.warnings.append({
                                'line': line_num,
                                'type': 'not_object',
                                'message': f'Record is {type(record).__name__}, expected object/dict'
                            })

                    except json.JSONDecodeError as e:
                        self.stats['parse_errors'] += 1
                        self.errors.append({
                            'line': line_num,
                            'type': 'parse_error',
                            'message': str(e),
                            'content': line[:100]  # First 100 chars
                        })

        except UnicodeDecodeError as e:
            self.errors.append({
                'line': 0,
                'type': 'encoding_error',
                'message': f'File encoding error: {e}'
            })

        return self.get_report()

    def get_report(self) -> Dict:
        """Generate validation report"""
        return {
            'valid': len(self.errors) == 0,
            'stats': self.stats,
            'errors': self.errors,
            'warnings': self.warnings
        }

    def print_report(self):
        """Print human-readable report"""
        report = self.get_report()

        print(f"\n{'='*60}")
        print(f"JSONL Validation Report: {self.filepath}")
        print(f"{'='*60}\n")

        print("Statistics:")
        print(f"  Total lines:    {self.stats['total_lines']:,}")
        print(f"  Valid records:  {self.stats['valid_records']:,}")
        print(f"  Empty lines:    {self.stats['empty_lines']:,}")
        print(f"  Parse errors:   {self.stats['parse_errors']:,}")
        print(f"  File size:      {self.stats['total_bytes']:,} bytes")

        if self.errors:
            print(f"\n{len(self.errors)} ERRORS:")
            for error in self.errors[:10]:  # Show first 10
                print(f"\n  Line {error['line']}: {error['type']}")
                print(f"    {error['message']}")
                if 'content' in error:
                    print(f"    Content: {error['content']}")

            if len(self.errors) > 10:
                print(f"\n  ... and {len(self.errors) - 10} more errors")

        if self.warnings:
            print(f"\n{len(self.warnings)} WARNINGS:")
            for warning in self.warnings[:10]:
                print(f"  Line {warning['line']}: {warning['message']}")

            if len(self.warnings) > 10:
                print(f"  ... and {len(self.warnings) - 10} more warnings")

        if report['valid']:
            print("\n✓ File is valid JSONL")
        else:
            print("\n✗ File has errors")

        return report['valid']

# Usage
if __name__ == '__main__':
    if len(sys.argv) < 2:
        print("Usage: python validator.py <file.jsonl>")
        sys.exit(1)

    validator = JSONLValidator(sys.argv[1])
    validator.validate()
    is_valid = validator.print_report()

    sys.exit(0 if is_valid else 1)

Usage: python validator.py data.jsonl

Command-Line Validation 

# Validate with jq (each line must be valid JSON)
cat data.jsonl | jq -c . > /dev/null && echo "Valid JSONL" || echo "Invalid JSONL"

# Show line numbers of invalid JSON
awk '{print NR, $0}' data.jsonl | while read -r num line; do
    echo "$line" | jq . > /dev/null 2>&1 || echo "Error on line $num"
done

# Count valid vs invalid lines
total=$(wc -l < data.jsonl)
valid=$(cat data.jsonl | jq -c . 2>/dev/null | wc -l)
echo "Valid: $valid / $total"

# Find lines with common issues
grep -n '^\s' data.jsonl  # Lines with leading whitespace
grep -n '\s$' data.jsonl  # Lines with trailing whitespace
grep -n '^$' data.jsonl   # Empty lines

# Validate and show specific errors
cat data.jsonl | while IFS= read -r line; do
    echo "$line" | jq . > /dev/null 2>&1 || echo "Error: $line"
done

Schema Validation 

from jsonschema import validate, ValidationError
import json

# Define expected schema
SCHEMA = {
    "type": "object",
    "required": ["id", "name", "email"],
    "properties": {
        "id": {"type": "integer", "minimum": 1},
        "name": {"type": "string", "minLength": 1},
        "email": {"type": "string", "format": "email"},
        "age": {"type": "integer", "minimum": 0, "maximum": 150}
    }
}

def validate_jsonl_schema(filepath, schema):
    """Validate records against schema"""
    errors = []

    with open(filepath, 'r') as f:
        for line_num, line in enumerate(f, 1):
            try:
                record = json.loads(line)

                # Validate against schema
                validate(instance=record, schema=schema)

            except json.JSONDecodeError as e:
                errors.append({
                    'line': line_num,
                    'error': 'JSON parse error',
                    'message': str(e)
                })

            except ValidationError as e:
                errors.append({
                    'line': line_num,
                    'error': 'Schema validation failed',
                    'message': e.message,
                    'path': list(e.path)
                })

    return errors

# Run validation
errors = validate_jsonl_schema('data.jsonl', SCHEMA)

if errors:
    print(f"Found {len(errors)} validation errors:")
    for error in errors[:20]:
        print(f"  Line {error['line']}: {error['message']}")
else:
    print("All records match schema")

Encoding Issues

Common Encoding Problems

1. BOM (Byte Order Mark)

UTF-8 BOM at start of file causes JSON parse errors

# Detect BOM
hexdump -C data.jsonl | head -n 1
# Look for: ef bb bf (UTF-8 BOM)

# Remove BOM
tail -c +4 data.jsonl > data_fixed.jsonl  # Skip first 3 bytes

# Python: Remove BOM
with open('data.jsonl', 'r', encoding='utf-8-sig') as f:
    content = f.read()
with open('data_fixed.jsonl', 'w', encoding='utf-8') as f:
    f.write(content)

2. Mixed Encoding

File contains mix of UTF-8 and Latin-1

import chardet

def detect_encoding(filepath):
    """Detect file encoding"""
    with open(filepath, 'rb') as f:
        result = chardet.detect(f.read(10000))
    return result

# Auto-detect and convert
def fix_encoding(input_file, output_file):
    """Convert to UTF-8"""
    encoding = detect_encoding(input_file)['encoding']
    print(f"Detected encoding: {encoding}")

    with open(input_file, 'r', encoding=encoding, errors='replace') as f_in:
        with open(output_file, 'w', encoding='utf-8') as f_out:
            for line in f_in:
                f_out.write(line)

fix_encoding('data.jsonl', 'data_utf8.jsonl')

3. Invalid UTF-8 Sequences

Corrupted characters in file

def clean_invalid_utf8(input_file, output_file):
    """Remove or replace invalid UTF-8"""
    with open(input_file, 'rb') as f_in:
        content = f_in.read()

    # Replace invalid sequences
    cleaned = content.decode('utf-8', errors='replace')

    with open(output_file, 'w', encoding='utf-8') as f_out:
        f_out.write(cleaned)

# Alternative: ignore invalid characters
with open('data.jsonl', 'r', encoding='utf-8', errors='ignore') as f:
    for line in f:
        process(line)

4. Line Ending Issues

Windows (CRLF) vs Unix (LF) line endings

# Convert Windows to Unix
dos2unix data.jsonl

# Or with Python
def fix_line_endings(input_file, output_file):
    """Convert to Unix line endings"""
    with open(input_file, 'rb') as f_in:
        content = f_in.read()

    # Replace CRLF with LF
    content = content.replace(b'\r\n', b'\n')

    with open(output_file, 'wb') as f_out:
        f_out.write(content)

Malformed JSON Detection

Find and Fix Common Issues 

import json
import re

def find_malformed_records(filepath):
    """Identify and categorize malformed JSON"""
    issues = {
        'missing_quotes': [],
        'trailing_commas': [],
        'unescaped_quotes': [],
        'incomplete_json': [],
        'other': []
    }

    with open(filepath, 'r', encoding='utf-8', errors='replace') as f:
        for line_num, line in enumerate(f, 1):
            try:
                json.loads(line)
            except json.JSONDecodeError as e:
                content = line.strip()

                # Categorize error
                if 'Expecting property name' in str(e):
                    issues['missing_quotes'].append((line_num, content, str(e)))
                elif 'trailing comma' in str(e).lower():
                    issues['trailing_commas'].append((line_num, content, str(e)))
                elif 'Unterminated string' in str(e):
                    issues['unescaped_quotes'].append((line_num, content, str(e)))
                elif 'Expecting value' in str(e):
                    issues['incomplete_json'].append((line_num, content, str(e)))
                else:
                    issues['other'].append((line_num, content, str(e)))

    return issues

# Auto-fix common issues
def auto_fix_jsonl(input_file, output_file):
    """Attempt to fix common JSON issues"""
    fixed = 0
    skipped = 0

    with open(input_file, 'r', encoding='utf-8', errors='replace') as f_in:
        with open(output_file, 'w', encoding='utf-8') as f_out:
            for line in f_in:
                try:
                    # Try parsing as-is
                    record = json.loads(line)
                    f_out.write(json.dumps(record) + '\n')
                    fixed += 1

                except json.JSONDecodeError:
                    # Try common fixes
                    fixed_line = line.strip()

                    # Remove trailing commas
                    fixed_line = re.sub(r',(\s*[}\]])', r'\1', fixed_line)

                    # Try parsing again
                    try:
                        record = json.loads(fixed_line)
                        f_out.write(json.dumps(record) + '\n')
                        fixed += 1
                    except json.JSONDecodeError:
                        skipped += 1
                        print(f"Could not fix line: {line[:100]}")

    print(f"Fixed: {fixed}, Skipped: {skipped}")
    return fixed, skipped

# Report malformed records
issues = find_malformed_records('data.jsonl')

for issue_type, records in issues.items():
    if records:
        print(f"\n{issue_type.upper()}: {len(records)} records")
        for line_num, content, error in records[:5]:
            print(f"  Line {line_num}: {error}")
            print(f"    {content[:100]}")

# Attempt auto-fix
auto_fix_jsonl('data.jsonl', 'data_fixed.jsonl')

Common Malformed Patterns

Issue Example Fix
Trailing comma {"a": 1, "b": 2,} {"a": 1, "b": 2}
Unquoted keys {name: "Alice"} {"name": "Alice"}
Single quotes {'name': 'Alice'} {"name": "Alice"}
Unescaped quotes {"text": "He said "hi""} {"text": "He said \"hi\""}
Incomplete JSON {"name": "Alice" {"name": "Alice"}
Multiple records per line {"a": 1}{"b": 2} Split into separate lines

Parser Error Diagnosis

Debug Parser Errors 

import json
import traceback

def debug_parse_error(line, line_num):
    """Detailed parser error analysis"""
    try:
        json.loads(line)
        return None

    except json.JSONDecodeError as e:
        error_info = {
            'line_num': line_num,
            'error_type': type(e).__name__,
            'message': str(e),
            'position': e.pos,
            'line_content': line,
            'error_location': line[max(0, e.pos-20):min(len(line), e.pos+20)]
        }

        # Show context around error
        if e.pos < len(line):
            error_info['char_at_error'] = repr(line[e.pos])

        return error_info

# Analyze all errors in file
def analyze_all_errors(filepath):
    """Comprehensive error analysis"""
    errors = []

    with open(filepath, 'r', encoding='utf-8', errors='replace') as f:
        for line_num, line in enumerate(f, 1):
            error = debug_parse_error(line, line_num)
            if error:
                errors.append(error)

    # Categorize errors
    error_types = {}
    for error in errors:
        msg = error['message']
        error_types[msg] = error_types.get(msg, 0) + 1

    print(f"Found {len(errors)} parse errors\n")
    print("Error type distribution:")
    for msg, count in sorted(error_types.items(), key=lambda x: -x[1]):
        print(f"  {count:4d} - {msg}")

    print("\nFirst 5 errors with context:")
    for error in errors[:5]:
        print(f"\nLine {error['line_num']}: {error['message']}")
        print(f"  Position: {error['position']}")
        print(f"  Context: ...{error['error_location']}...")
        if 'char_at_error' in error:
            print(f"  Character at error: {error['char_at_error']}")

analyze_all_errors('data.jsonl')

Extract Valid Records 

def extract_valid_records(input_file, output_file, error_file=None):
    """Separate valid and invalid records"""
    valid_count = 0
    error_count = 0

    with open(input_file, 'r', encoding='utf-8', errors='replace') as f_in:
        with open(output_file, 'w', encoding='utf-8') as f_out:
            error_writer = None
            if error_file:
                error_writer = open(error_file, 'w', encoding='utf-8')

            for line_num, line in enumerate(f_in, 1):
                try:
                    record = json.loads(line)
                    f_out.write(json.dumps(record) + '\n')
                    valid_count += 1

                except json.JSONDecodeError as e:
                    error_count += 1
                    if error_writer:
                        error_writer.write(f"Line {line_num}: {e}\n")
                        error_writer.write(line)
                        error_writer.write('\n')

            if error_writer:
                error_writer.close()

    print(f"Valid: {valid_count}, Errors: {error_count}")
    return valid_count, error_count

# Usage
extract_valid_records('data.jsonl', 'valid.jsonl', 'errors.txt')

Memory Issues

Diagnose Memory Problems 

import tracemalloc
import json

def profile_memory_usage(filepath):
    """Profile memory usage while processing"""
    tracemalloc.start()

    # Snapshot before
    snapshot1 = tracemalloc.take_snapshot()

    # Bad: Load entire file
    with open(filepath, 'r') as f:
        data = [json.loads(line) for line in f]

    # Snapshot after
    snapshot2 = tracemalloc.take_snapshot()

    # Compare
    top_stats = snapshot2.compare_to(snapshot1, 'lineno')

    print("Memory usage (loading entire file):")
    for stat in top_stats[:10]:
        print(stat)

    current, peak = tracemalloc.get_traced_memory()
    print(f"\nCurrent memory: {current / 1024 / 1024:.1f} MB")
    print(f"Peak memory: {peak / 1024 / 1024:.1f} MB")

    tracemalloc.stop()

# Good: Stream processing
def stream_with_profiling(filepath):
    """Stream processing with memory monitoring"""
    tracemalloc.start()

    record_count = 0
    for line in open(filepath, 'r'):
        record = json.loads(line)
        record_count += 1

        if record_count % 10000 == 0:
            current, peak = tracemalloc.get_traced_memory()
            print(f"Processed {record_count}: {current / 1024 / 1024:.1f} MB")

    tracemalloc.stop()

profile_memory_usage('large_file.jsonl')
stream_with_profiling('large_file.jsonl')

Fix Memory Leaks 

# Bad: Accumulates records in memory
def process_bad(filepath):
    results = []
    with open(filepath, 'r') as f:
        for line in f:
            record = json.loads(line)
            processed = transform(record)
            results.append(processed)  # Memory leak!
    return results

# Good: Stream processing
def process_good(filepath, output_file):
    with open(filepath, 'r') as f_in:
        with open(output_file, 'w') as f_out:
            for line in f_in:
                record = json.loads(line)
                processed = transform(record)
                f_out.write(json.dumps(processed) + '\n')
                # Record immediately discarded after writing

# Good: Generator
def process_generator(filepath):
    with open(filepath, 'r') as f:
        for line in f:
            record = json.loads(line)
            yield transform(record)

# Use generator
for result in process_generator('data.jsonl'):
    handle_result(result)  # Process one at a time

Performance Troubleshooting

Benchmark & Profile 

import time
import cProfile
import pstats

def benchmark_processing(filepath):
    """Benchmark different processing methods"""

    # Method 1: Load all
    start = time.time()
    with open(filepath, 'r') as f:
        data = [json.loads(line) for line in f]
    method1_time = time.time() - start

    # Method 2: Stream
    start = time.time()
    count = 0
    for line in open(filepath, 'r'):
        record = json.loads(line)
        count += 1
    method2_time = time.time() - start

    # Method 3: Streaming with orjson (faster JSON parser)
    import orjson
    start = time.time()
    count = 0
    for line in open(filepath, 'rb'):
        record = orjson.loads(line)
        count += 1
    method3_time = time.time() - start

    print(f"Load all: {method1_time:.2f}s")
    print(f"Stream (json): {method2_time:.2f}s")
    print(f"Stream (orjson): {method3_time:.2f}s")
    print(f"orjson is {method2_time / method3_time:.1f}x faster")

# Profile specific function
def profile_function(filepath):
    """Profile with cProfile"""

    def process_file():
        for line in open(filepath, 'r'):
            record = json.loads(line)
            # Process...

    profiler = cProfile.Profile()
    profiler.enable()

    process_file()

    profiler.disable()
    stats = pstats.Stats(profiler)
    stats.sort_stats('cumulative')
    stats.print_stats(20)  # Top 20 functions

benchmark_processing('large_file.jsonl')
profile_function('large_file.jsonl')

Optimization Strategies

1. Use Faster JSON Parser

# Install: pip install orjson
# ujson is no longer recommended - it is effectively deprecated.
# msgspec is another fast option if you also want typed decoding.
import orjson  # Fastest
import json    # Standard library, fine for most workloads

# Benchmark
import time

# Standard json
start = time.time()
for line in open('data.jsonl', 'r'):
    json.loads(line)
json_time = time.time() - start

# orjson (binary mode)
start = time.time()
for line in open('data.jsonl', 'rb'):
    orjson.loads(line)
orjson_time = time.time() - start

print(f"json: {json_time:.2f}s")
print(f"orjson: {orjson_time:.2f}s ({json_time/orjson_time:.1f}x faster)")

2. Parallel Processing

from multiprocessing import Pool
import json

def process_chunk(lines):
    return [json.loads(line) for line in lines]

def parallel_process(filepath, num_workers=4):
    with open(filepath, 'r') as f:
        lines = f.readlines()

    # Split into chunks
    chunk_size = len(lines) // num_workers
    chunks = [lines[i:i+chunk_size] for i in range(0, len(lines), chunk_size)]

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

    return [item for sublist in results for item in sublist]

3. Buffered I/O

# Default buffering (8KB)
with open('data.jsonl', 'r') as f:
    for line in f:
        process(line)

# Larger buffer (1MB)
with open('data.jsonl', 'r', buffering=1024*1024) as f:
    for line in f:
        process(line)

# Read in chunks
def read_chunks(filepath, chunk_size=1024*1024):
    with open(filepath, 'r') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

Data Quality Checks

Comprehensive Quality Analysis 

import json
from collections import Counter, defaultdict

class DataQualityAnalyzer:
    """Analyze JSONL data quality"""

    def __init__(self, filepath):
        self.filepath = filepath
        self.stats = {
            'total_records': 0,
            'unique_keys': set(),
            'missing_keys': defaultdict(int),
            'null_values': defaultdict(int),
            'empty_strings': defaultdict(int),
            'data_types': defaultdict(Counter),
            'duplicates': 0
        }
        self.seen_records = set()

    def analyze(self):
        """Run full analysis"""
        with open(self.filepath, 'r') as f:
            for line in f:
                try:
                    record = json.loads(line)
                    self.analyze_record(record)
                except json.JSONDecodeError:
                    continue

        return self.get_report()

    def analyze_record(self, record):
        """Analyze single record"""
        self.stats['total_records'] += 1

        # Check for duplicates
        record_hash = json.dumps(record, sort_keys=True)
        if record_hash in self.seen_records:
            self.stats['duplicates'] += 1
        self.seen_records.add(record_hash)

        # Collect keys
        self.stats['unique_keys'].update(record.keys())

        # Check each field
        for key, value in record.items():
            # Data type distribution
            self.stats['data_types'][key][type(value).__name__] += 1

            # Null values
            if value is None:
                self.stats['null_values'][key] += 1

            # Empty strings
            if value == '':
                self.stats['empty_strings'][key] += 1

    def get_report(self):
        """Generate quality report"""
        report = {
            'total_records': self.stats['total_records'],
            'unique_keys': len(self.stats['unique_keys']),
            'duplicates': self.stats['duplicates'],
            'keys': list(self.stats['unique_keys']),
            'quality_issues': []
        }

        # Find keys with high null rate
        for key, null_count in self.stats['null_values'].items():
            null_rate = null_count / self.stats['total_records']
            if null_rate > 0.1:  # >10% null
                report['quality_issues'].append({
                    'key': key,
                    'issue': 'high_null_rate',
                    'rate': f"{null_rate*100:.1f}%"
                })

        # Find keys with inconsistent types
        for key, type_counts in self.stats['data_types'].items():
            if len(type_counts) > 1:
                report['quality_issues'].append({
                    'key': key,
                    'issue': 'inconsistent_types',
                    'types': dict(type_counts)
                })

        return report

    def print_report(self):
        """Print human-readable report"""
        report = self.get_report()

        print(f"\nData Quality Report")
        print("=" * 60)
        print(f"Total records: {report['total_records']:,}")
        print(f"Unique keys: {report['unique_keys']}")
        print(f"Duplicates: {report['duplicates']}")

        if report['quality_issues']:
            print(f"\nQuality Issues ({len(report['quality_issues'])}):")
            for issue in report['quality_issues']:
                print(f"  {issue['key']}: {issue['issue']}")
                if 'rate' in issue:
                    print(f"    Rate: {issue['rate']}")
                if 'types' in issue:
                    print(f"    Types: {issue['types']}")

# Usage
analyzer = DataQualityAnalyzer('data.jsonl')
analyzer.analyze()
analyzer.print_report()

Debugging Tools & Scripts

Essential Shell Commands 

# Count records
wc -l data.jsonl

# Check file encoding
file -i data.jsonl

# Find records matching pattern
grep -n '"status": "error"' data.jsonl

# Extract specific field from all records
cat data.jsonl | jq -r '.email'

# Count unique values for a field
cat data.jsonl | jq -r '.category' | sort | uniq -c

# Find records with missing field
cat data.jsonl | jq 'select(.email == null)'

# Pretty print first record
head -n 1 data.jsonl | jq .

# Check for duplicate records
sort data.jsonl | uniq -d

# Split large file into chunks
split -l 1000000 data.jsonl chunk_

# Merge multiple JSONL files
cat file1.jsonl file2.jsonl > merged.jsonl

# Random sample of records
shuf -n 100 data.jsonl > sample.jsonl

Debug Utility Script 

#!/usr/bin/env python3
"""
JSONL Debug Utility
Usage: python jsonl_debug.py <file.jsonl> [--validate] [--stats] [--fix]
"""

import json
import sys
import argparse

def main():
    parser = argparse.ArgumentParser(description='JSONL debugging utility')
    parser.add_argument('file', help='JSONL file to analyze')
    parser.add_argument('--validate', action='store_true', help='Validate JSON syntax')
    parser.add_argument('--stats', action='store_true', help='Show statistics')
    parser.add_argument('--fix', help='Attempt to fix and save to file')
    parser.add_argument('--encoding', action='store_true', help='Check encoding')

    args = parser.parse_args()

    if args.validate:
        validate_file(args.file)

    if args.stats:
        show_stats(args.file)

    if args.fix:
        fix_file(args.file, args.fix)

    if args.encoding:
        check_encoding(args.file)

if __name__ == '__main__':
    main()

Troubleshooting Flowchart

Problem: Cannot parse JSONL file

1. Check file encoding

  • Run: file -i data.jsonl
  • Look for BOM: hexdump -C data.jsonl | head -n 1
  • If not UTF-8, convert with encoding detection script

2. Validate JSON syntax

  • Run: cat data.jsonl | jq -c . > /dev/null
  • Use Python validator to find specific errors
  • Check for trailing commas, unquoted keys, single quotes

3. Check for empty/whitespace lines

  • Find: grep -n '^$' data.jsonl
  • Remove: sed '/^$/d' data.jsonl > cleaned.jsonl

4. Memory or performance issues

  • Use streaming instead of loading entire file
  • Profile with tracemalloc
  • Consider parallel processing for large files

If all steps pass: File is valid JSONL

Related Resources

Advanced Topics

Compression and optimization

Performance Guide

Speed optimization

Best Practices

Production patterns

Examples

Code samples

Error Dictionary

Parser messages decoded

Glossary

Terms used in this guide

Tools

Libraries and utilities

External Links

Community resources