Como fazer parsing de ficheiros JSONL em ETL: passo a passo

Este tutorial mostra como fazer parsing de ficheiros JSONL em ETL e transformar registos JSON por linha para um formato tabular útil em análises. É útil quando os dados chegam em JSONL (logs, eventos) e é necessário normalizar e carregar para uma base de dados ou CSV.

Pré-requisitos

• Python 3.8+ instalado
• Bibliotecas: pandas, fastjsonschema (opcional para validação)
• Ficheiro JSONL de exemplo (cada linha é um objeto JSON)
• Editor de texto ou IDE e linha de comandos

Passo 1: perceber o formato JSONL e definir o objetivo

JSONL (JSON Lines) tem um registo JSON por linha. Primeiro, abre algumas linhas para verificar campos simples e aninhados. Decide quais colunas queres extrair e se precisas de normalizar arrays ou objetos aninhados.

Passo 2: ler JSONL em streaming para evitar esgotamento de memória

Para ficheiros grandes lê em streaming linha a linha e processa por blocos. Assim evitas carregar tudo na memória. Este exemplo lê 10.000 linhas por bloco e transforma em DataFrame para operações em lote.

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]

Passo 3: normalizar campos aninhados e arrays

Usa pandas.json_normalize para desanexar objetos aninhados. Para arrays, decide se explodes (uma linha por elemento) ou agregas (concatenar/contar). No exemplo abaixo normalizamos campos "user" e explodimos o array "events".

from pandas import json_normalize

def process_chunk(objs):
    # transforma lista de dicts num DataFrame plano
    df = json_normalize(objs)
    # exemplo: user.id -> user.id, user.name -> user.name
    # lidar com array events: cada registo pode ter lista events
    if 'events' in df.columns:
        df_events = df.explode('events')
        # events é agora dict ou NaN
        events_df = json_normalize(df_events['events'].dropna())
        # reindex para alinhar com 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

Passo 4: validação e tratamento de erros comuns

Erros comuns: linhas malformadas, campos em formatos inesperados e valores nulos. Valida cada linha com try/except e regista as falhas para análise posterior. Podes usar fastjsonschema para validar a estrutura antes de processar.

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

# exemplo de uso com 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 com transformações / persistência

Passo 5: normalizar tipos, tratar timestamps e chaves primárias

Converte timestamps para datetime, normaliza strings e gera uma chave primária consistente (hash) se não existirem ids. Isto ajuda na carga incremental e deduplicação.

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:
        # gera id com 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

Passo 6: gravar em CSV ou carregar para a base de dados em blocos

Grava em CSV por blocos ou usa um conector para a tua base de dados. Ao gravar em CSV usa header na primeira escrita e append sem header nos blocos seguintes.

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 o resultado

Abre o CSV resultante ou consulta a tabela da base de dados e verifica: número de linhas coerente, colunas esperadas, timestamps convertidos e ausência de erros previstos. Verifica também uma amostra de registos com eventos explodidos para confirmar o alinhamento.

Conclusão

Com estas etapas já tens um fluxo ETL prático para fazer parsing de ficheiros JSONL, normalizar estruturas e carregar para análise. Próximos passos: adicionar testes, validação por esquema com fastjsonschema e suporte a compressão (gzip). Dica: cria amostras pequenas e valida antes de processar ficheiros grandes para evitar surpresas.