Cómo hacer parsing de archivos JSONL en ETL: paso a paso

Este tutorial muestra cómo hacer parsing de archivos JSONL en ETL y transformar registros JSON por línea a un formato tabular útil en análisis. Es útil cuando los datos llegan en JSONL (logs, eventos) y es necesario normalizar y cargar en una base de datos o CSV.

Requisitos previos

• Python 3.8+ instalado
• Bibliotecas: pandas, fastjsonschema (opcional para validación)
• Archivo JSONL de ejemplo (cada línea es un objeto JSON)
• Editor de texto o IDE y línea de comandos

Paso 1: entender el formato JSONL y definir el objetivo

JSONL (JSON Lines) tiene un registro JSON por línea. Primero, abre algunas líneas para verificar campos simples y anidados. Decide qué columnas quieres extraer y si necesitas normalizar arrays u objetos anidados.

Paso 2: leer JSONL en streaming para evitar agotamiento de memoria

Para archivos grandes lee en streaming línea a línea y procesa por bloques. Así evitas cargar todo en memoria. Este ejemplo lee 10.000 líneas por bloque y transforma en DataFrame para operaciones por lotes.

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]

Paso 3: normalizar campos anidados y arrays

Usa pandas.json_normalize para desanidar objetos anidados. Para arrays, decide si explodes (una línea por elemento) o agregas (concatenar/contar). En el ejemplo abajo normalizamos campos "user" y explotamos el array "events".

from pandas import json_normalize

def process_chunk(objs):
    # transforma lista de dicts en un DataFrame plano
    df = json_normalize(objs)
    # ejemplo: user.id -> user.id, user.name -> user.name
    # manejar array events: cada registro puede tener lista events
    if 'events' in df.columns:
        df_events = df.explode('events')
        # events es ahora dict o NaN
        events_df = json_normalize(df_events['events'].dropna())
        # reindexar para alinear con 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

Paso 4: validación y tratamiento de errores comunes

Errores comunes: líneas malformadas, campos en formatos inesperados y valores nulos. Valida cada línea con try/except y registra las fallas para análisis posterior. Puedes usar fastjsonschema para validar la estructura antes de procesar.

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

# ejemplo de uso con 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)
    # seguir con transformaciones / persistencia

Paso 5: normalizar tipos, tratar timestamps y claves primarias

Convierte timestamps a datetime, normaliza strings y genera una clave primaria consistente (hash) si no existen ids. Esto ayuda en la carga incremental y deduplicación.

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:
        # genera id con hash de campos relevantes
        df['id'] = df.apply(lambda r: hashlib.sha1((''.join(map(str, [r.get('user.id'), r.get('timestamp')] ))).encode()).hexdigest(), axis=1)
    return df

Paso 6: escribir en CSV o cargar a la base de datos en bloques

Escribe en CSV por bloques o usa un conector para tu base de datos. Al escribir en CSV usa header en la primera escritura y append sin header en los bloques siguientes.

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')

Verificar el resultado

Abre el CSV resultante o consulta la tabla de la base de datos y verifica: número de filas coherente, columnas esperadas, timestamps convertidos y ausencia de errores previstos. Verifica también una muestra de registros con eventos explotados para confirmar el alineamiento.

Conclusión

Con estos pasos ya tienes un flujo ETL práctico para hacer parsing de archivos JSONL, normalizar estructuras y cargar para análisis. Próximos pasos: añadir tests, validación por esquema con fastjsonschema y soporte a compresión (gzip). Consejo: crea muestras pequeñas y valida antes de procesar archivos grandes para evitar sorpresas.