How to parse JSONL files in ETL: step by step

This tutorial shows how to parse JSONL files in ETL and transform JSON records per line into a tabular format useful for analysis. It is useful when data arrives in JSONL (logs, events) and you need to normalize and load into a database or CSV.

Prerequisites

• Python 3.8+ installed
• Libraries: pandas, fastjsonschema (optional for validation)
• Sample JSONL file (each line is a JSON object)
• Text editor or IDE and command line

Step 1: understand the JSONL format and define the goal

JSONL (JSON Lines) has one JSON record per line. First, open a few lines to inspect simple and nested fields. Decide which columns you want to extract and whether you need to normalize arrays or nested objects.

Step 2: read JSONL in streaming to avoid running out of memory

For large files read in streaming line by line and process in chunks. This avoids loading everything into memory. This example reads 10,000 lines per chunk and converts to a DataFrame for batch operations.

import json
import pandas as pd
from itertools import islice

def read_jsonl_in_chunks(path, chunk_size=10000):
    with open(path, 'r', encoding='utf-8') as f:
        while True:
            lines = list(islice(f, chunk_size))
            if not lines:
                break
            yield [json.loads(l) for l in lines]

Step 3: normalize nested fields and arrays

Use pandas.json_normalize to flatten nested objects. For arrays, decide whether to explode (one row per element) or aggregate (concatenate/count). In the example below we normalize "user" fields and explode the "events" array.

from pandas import json_normalize

def process_chunk(objs):
    # transform list of dicts into a flat DataFrame
    df = json_normalize(objs)
    # example: user.id -> user.id, user.name -> user.name
    # handle events array: each record may have an events list
    if 'events' in df.columns:
        df_events = df.explode('events')
        # events is now dict or NaN
        events_df = json_normalize(df_events['events'].dropna())
        # reindex to align with df_events
        events_df.index = df_events.index[:len(events_df)]
        df_events = df_events.drop(columns=['events']).join(events_df)
        return df_events
    return df

Step 4: validation and handling common errors

Common errors: malformed lines, fields in unexpected formats and null values. Validate each line with try/except and log failures for later analysis. You can use fastjsonschema to validate the structure before processing.

import logging
logging.basicConfig(level=logging.INFO)

def safe_load(line):
    try:
        return json.loads(line)
    except json.JSONDecodeError as e:
        logging.warning(f'Linha inválida: {e}')
        return None

# example usage with streaming
for chunk in read_jsonl_in_chunks('data.jsonl'):
    objs = [o for o in chunk if o is not None]
    df = process_chunk(objs)
    # continue with transformations / persistence

Step 5: normalize types, handle timestamps and primary keys

Convert timestamps to datetime, normalize strings and generate a consistent primary key (hash) if ids do not exist. This helps with incremental loading and deduplication.

import hashlib

def ensure_types(df):
    if 'timestamp' in df.columns:
        df['timestamp'] = pd.to_datetime(df['timestamp'], errors='coerce')
    if 'id' not in df.columns:
        # generate id with hash of relevant fields
        df['id'] = df.apply(lambda r: hashlib.sha1((''.join(map(str, [r.get('user.id'), r.get('timestamp')] ))).encode()).hexdigest(), axis=1)
    return df

Step 6: write to CSV or load into the database in chunks

Write to CSV in chunks or use a connector to your database. When writing to CSV use header on the first write and append without header for subsequent chunks.

output_csv = 'out.csv'
first = True
for chunk in read_jsonl_in_chunks('data.jsonl'):
    objs = [o for o in chunk if o is not None]
    df = process_chunk(objs)
    df = ensure_types(df)
    if first:
        df.to_csv(output_csv, index=False, mode='w', encoding='utf-8')
        first = False
    else:
        df.to_csv(output_csv, index=False, mode='a', header=False, encoding='utf-8')

Verify the result

Open the resulting CSV or query the database table and verify: coherent row count, expected columns, converted timestamps and absence of known errors. Also check a sample of records with exploded events to confirm alignment.

Conclusion

With these steps you have a practical ETL flow to parse JSONL files, normalize structures and load for analysis. Next steps: add tests, schema validation with fastjsonschema and support for compression (gzip). Tip: create small samples and validate before processing large files to avoid surprises.