Como validar a qualidade dos dados num pipeline ETL

Um pipeline ETL que carrega dados errados é pior do que não ter pipeline nenhum: os erros propagam-se em silêncio até aos relatórios e às decisões. Emails vazios, idades impossíveis ou clientes repetidos entram na tabela final sem ninguém dar por isso. A validação da qualidade dos dados resolve o problema ao verificar cada linha antes da carga, deixando entrar apenas as que cumprem as tuas regras. É um passo simples de acrescentar com Python e pandas, e faz toda a diferença na confiança que tens nos teus dados.

Pré-requisitos

• Python 3.9 ou superior instalado no teu computador.
• A biblioteca pandas instalada (pip install pandas).
• Um ficheiro CSV de exemplo para validar (usamos clientes.csv).
• Noções básicas do fluxo ETL: extrair, transformar e carregar.

Passo 1: Extrair os dados e definir as regras

O primeiro passo é ler os dados de origem para um DataFrame e escrever, de forma explícita, as regras que cada linha tem de respeitar. Ter as regras reunidas num único sítio torna o pipeline fácil de manter e de auditar: se amanhã a regra da idade mudar, alteras um único número. No nosso exemplo temos uma tabela de clientes e quatro regras de qualidade: o email não pode estar vazio, a idade tem de estar entre 18 e 120, o país tem de pertencer a uma lista permitida e o id tem de ser único.

import pandas as pd

# Extrair (E): ler os dados de origem
df = pd.read_csv("clientes.csv")

# Lista de valores permitidos para o pais
PAISES_VALIDOS = ["PT", "ES", "FR", "DE"]

Passo 2: Escrever a função de validação

Em vez de apagar logo as linhas inválidas, vamos anotar o motivo de cada falha. Assim guardamos um registo claro do que correu mal, o que ajuda a corrigir a origem mais tarde. A função a seguir recebe uma linha e devolve uma lista com todos os problemas encontrados — repara que juntamos todos os erros, e não apenas o primeiro. Desta forma, quem corrige os dados vê todos os problemas de uma vez em vez de os descobrir um a um.

def validar_linha(linha):
    problemas = []
    if pd.isna(linha["email"]) or str(linha["email"]).strip() == "":
        problemas.append("email vazio")
    if pd.isna(linha["idade"]) or linha["idade"] < 18 or linha["idade"] > 120:
        problemas.append("idade invalida")
    if linha["pais"] not in PAISES_VALIDOS:
        problemas.append("pais nao permitido")
    return problemas

Passo 3: Aplicar as regras e separar as linhas

Agora aplicamos a função a todas as linhas com apply e axis=1 e juntamos a verificação de ids duplicados. Ao contrário das outras regras, detetar duplicados obriga a comparar cada id com todos os outros, por isso corre sobre a coluna completa e não linha a linha. O resultado é uma coluna erros com o texto dos problemas de cada linha. As linhas sem erros são as válidas; as restantes vão para quarentena.

# Erros por linha (regras individuais)
df["erros"] = df.apply(validar_linha, axis=1)

# Regra que envolve a coluna toda: id duplicado
duplicados = df["id"].duplicated(keep=False)
df.loc[duplicados, "erros"] = df.loc[duplicados, "erros"].apply(
    lambda lista: lista + ["id duplicado"]
)

# Transformar a lista de erros em texto
df["erros"] = df["erros"].apply(lambda lista: ", ".join(lista))

# Separar validas de invalidas
validas = df[df["erros"] == ""].copy()
invalidas = df[df["erros"] != ""].copy()

Passo 4: Carregar as válidas e guardar as inválidas

Na fase de carga (L) enviamos apenas as linhas válidas para o destino. As inválidas não são deitadas fora: são gravadas num ficheiro de quarentena com a coluna erros, para que alguém as possa rever e corrigir. Aqui usamos ficheiros CSV para manter o exemplo simples, mas o destino podia ser uma base de dados ou um data warehouse.

# Carregar (L): so as linhas validas
validas.drop(columns=["erros"]).to_csv("clientes_limpos.csv", index=False)

# Quarentena: linhas invalidas com o motivo
invalidas.to_csv("clientes_quarentena.csv", index=False)

print(f"Validas: {len(validas)} | Invalidas: {len(invalidas)}")

Verificar o resultado

Para confirmar que a validação funcionou, abre o ficheiro clientes_quarentena.csv e vê se cada linha rejeitada tem uma explicação na coluna erros. O número de linhas válidas mais o número de inválidas deve ser igual ao total lido no início — a linha assert confirma isso automaticamente e falha se as contas não fecharem.

# Verificacao rapida: as contas fecham?
assert len(validas) + len(invalidas) == len(df)
print(invalidas[["id", "erros"]])

Conclusão

Com poucas linhas de código passaste a ter um pipeline ETL que confia, mas verifica: as linhas boas seguem para o destino e as duvidosas ficam em quarentena com o motivo. A partir daqui podes acrescentar novas regras (formato do email, datas no futuro, valores negativos), transformar esta verificação numa etapa reutilizável ou registar métricas de qualidade a cada execução. Qual é a regra de qualidade mais importante para os teus dados — e já a estás a validar?