Qualitat de dades

Per qué importa la qualitat de dades

"Garbage in, garbage out" és el principi més vell de la informàtica aplicada a les dades: si alimentes un model analític o un pipeline amb dades incorrectes, les conclusions que n'obtens seran incorrectes. El problema és que les dades incorrectes no generen errors evidents; generen respostes plausibles però falses.

El cost de detectar un error de qualitat augmenta exponencialment com més tard es detecta:

• Detectat en el pipeline d'entrada: cost zero (es rebutja el registre).
• Detectat al Data Warehouse: cal una correcció, possiblement un backfill.
• Detectat al dashboard que llegeix el CEO: cal corregir, recalcular, comunicar, i gestionar la pèrdua de confiança en el sistema.
• Detectat després de prendre decisions de negoci equivocades: cost màxim, pot ser irreversible.

Casos reals documentats: el 2012, Knight Capital Group va perdre 440 milions de dòlars en 45 minuts per un error en un pipeline automatitzat. El 2020, el Regne Unit va perdre 16.000 casos de COVID-19 per un error en un fitxer Excel que havia superat el límit de files. La qualitat de dades no és una tasca administrativa; és infraestructura crítica.

---

Dimensions de qualitat de dades

Completitud

Mesura si tots els valors obligatoris estan presents. Un camp email buit en una taula de clients impedeix enviar comunicacions. Un camp preu nul en una taula de transaccions fa impossible calcular ingressos.

import pandas as pd

df = pd.read_csv("clients.csv")

# Percentatge de completitud per columna
completitud = (1 - df.isnull().mean()) * 100
print(completitud.sort_values())

Exactitud

Els valors són correctes respecte a la realitat? Una edat negativa, un preu de -50 euros o un codi postal de 6 dígits (quan n'hauria de tenir 5) son exemples d'inexactitud.

# Regles d'exactitud
invalids_edat = df[~df["edat"].between(0, 120)]
invalids_preu = df[df["preu"] < 0]
invalids_cp   = df[~df["codi_postal"].astype(str).str.match(r"^\d{5}$")]

Consistència

La mateixa entitat apareix amb valors contradictoris en dos sistemes o en dues taules. El client amb id=1042 té pais="ES" al CRM però pais="España" al sistema de facturació. Ambdues fonts estan "correctes" localment, però son inconsistents entre elles.

Unicitat

No hi ha duplicats on no n'hi hauria d'haver. Si la taula de clients té la mateixa persona dues vegades (per un error d'importació), totes les agregacions per client estaran inflades.

# Detectar duplicats per clau primària
duplicats = df[df.duplicated(subset=["id_client"], keep=False)]
print(f"Registres duplicats: {len(duplicats)}")

Temporalitat

Les dades son actuals? Un timestamp amb data del 1970 (epoch zero) indica un error de conversió. Una comanda amb data_lliurament anterior a data_comanda és físicament impossible.

df["data_comanda"] = pd.to_datetime(df["data_comanda"])
df["data_lliurament"] = pd.to_datetime(df["data_lliurament"])

# Detectar lliuraments anteriors a la comanda
errors_temporals = df[df["data_lliurament"] < df["data_comanda"]]

Validesa

El format és correcte? Un camp email que no conté @, un camp IBAN amb longitud incorrecta o un camp data en format DD/MM/YYYY quan s'espera YYYY-MM-DD son errors de validesa.

---

Validació amb Pandera

Pandera permet definir un esquema de validació per a DataFrames de pandas de forma declarativa. Si el DataFrame no compleix l'esquema, Pandera llança una excepció amb els detalls dels errors:

import pandera as pa
from pandera import Column, DataFrameSchema, Check
import pandas as pd

# Definir l'esquema de validació
esquema_vendes = DataFrameSchema(
    columns={
        "id_venda": Column(
            dtype=int,
            checks=Check.greater_than(0),
            nullable=False,
            unique=True,
        ),
        "data": Column(
            dtype="datetime64[ns]",
            nullable=False,
        ),
        "id_client": Column(
            dtype=int,
            checks=Check.greater_than(0),
            nullable=False,
        ),
        "import": Column(
            dtype=float,
            checks=[
                Check.greater_than(0),
                Check.less_than(100_000),
            ],
            nullable=False,
        ),
        "email_client": Column(
            dtype=str,
            checks=Check.str_matches(r"^[^@]+@[^@]+\.[^@]+$"),
            nullable=True,   # l'email pot ser buit
        ),
        "pais": Column(
            dtype=str,
            checks=Check.isin(["ES", "FR", "DE", "IT", "PT"]),
            nullable=False,
        ),
    },
    coerce=True,       # intenta convertir els tipus automàticament
    strict=False,      # permet columnes addicionals no declarades
)

# Aplicar la validació
try:
    df_validat = esquema_vendes.validate(df, lazy=True)
    print(f"Validació correcta: {len(df_validat):,} registres")
except pa.errors.SchemaErrors as e:
    print("Errors de validació trobats:")
    print(e.failure_cases)
    # e.failure_cases és un DataFrame amb: columna, índex, valor invàlid, regla fallida

El paràmetre lazy=True fa que Pandera recopili tots els errors en lloc d'aturar-se al primer, cosa que permet tenir un informe complet dels problemes del dataset.

---

Validació amb Great Expectations

Great Expectations (GE) és la llibreria de Data Quality més completa per a Python. Permet definir "expectatives" sobre les dades, executar-les i generar un informe HTML detallat (Data Docs).

Conceptes principals

• Expectation: una afirmació sobre les dades que ha de ser certa. expect_column_values_to_not_be_null("email"), expect_column_values_to_be_between("edat", 0, 120).
• Expectation Suite: conjunt d'expectatives aplicades a un dataset.
• Checkpoint: combina una font de dades amb una suite d'expectatives i genera un resultat de validació.
• Data Docs: informe HTML generat automàticament que mostra quines expectatives han passat i quines han fallat.

Exemple complet de validació amb GE 0.18+

import great_expectations as gx
import pandas as pd

# Carregar el DataFrame a validar
df = pd.read_csv("/dades/vendes_2026_06.csv")

# Crear el context de GE (mode in-memory, sense fitxers de configuració)
context = gx.get_context(mode="ephemeral")

# Definir la font de dades (pandas)
font = context.data_sources.add_pandas("vendes_pandas")
actiu = font.add_dataframe_asset("vendes_juny")
lot = actiu.add_batch_definition_whole_dataframe("lot_complet")

# Crear la suite d'expectatives
suite = context.suites.add(gx.ExpectationSuite(name="suite_vendes"))

# Afegir expectatives
suite.add_expectation(
    gx.expectations.ExpectColumnToExist(column="id_venda")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="id_venda")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeUnique(column="id_venda")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="import",
        min_value=0.01,
        max_value=99999.99,
    )
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToMatchRegex(
        column="email_client",
        regex=r"^[^@]+@[^@]+\.[^@]+$",
        mostly=0.95,   # accepta fins a un 5% de valors invàlids
    )
)
suite.add_expectation(
    gx.expectations.ExpectTableRowCountToBeBetween(
        min_value=100,
        max_value=1_000_000,
    )
)

# Executar la validació
lot_executat = lot.get_batch(batch_parameters={"dataframe": df})
resultat = context.run_validations(
    validation_definition=context.validation_definitions.add(
        gx.ValidationDefinition(
            name="validacio_vendes",
            data=lot,
            suite=suite,
        )
    )
)

# Comprovar el resultat
if resultat.success:
    print("Totes les expectatives han passat.")
else:
    print("Hi ha expectatives fallides:")
    for res in resultat.results:
        if not res.success:
            print(f"  - {res.expectation_config.type}: {res.result}")
    raise RuntimeError("Validació fallida. Pipeline aturat.")

---

Perfil de dades (Data Profiling)

Abans de definir les expectatives, cal entendre el dataset. El data profiling genera estadístiques descriptives automàtiques:

# Profiling bàsic amb pandas (sense dependències extra)
def perfil_basic(df: pd.DataFrame) -> pd.DataFrame:
    resum = pd.DataFrame({
        "tipus": df.dtypes,
        "no_nuls": df.count(),
        "nuls": df.isnull().sum(),
        "pct_nuls": (df.isnull().mean() * 100).round(2),
        "únics": df.nunique(),
        "mostra": [df[c].dropna().iloc[0] if df[c].count() > 0 else None
                   for c in df.columns],
    })
    return resum

print(perfil_basic(df).to_string())

Per a reports HTML complets, la llibreria ydata-profiling (antiga pandas-profiling) genera un informe interactiu:

from ydata_profiling import ProfileReport

perfil = ProfileReport(df, title="Perfil vendes juny 2026", minimal=True)
perfil.to_file("perfil_vendes.html")

---

Gestió d'errors: quarantena i alertes

Un pipeline robust no s'atura davant d'un registre invàlid; l'envia a una zona de quarantena per a revisió:

def processa_amb_quarantena(
    df: pd.DataFrame,
    esquema: pa.DataFrameSchema,
    taula_quarantena: str,
    conn,
) -> pd.DataFrame:
    """Separa registres vàlids de invàlids i guarda la quarantena."""
    valids = []
    invalids = []

    for idx, fila in df.iterrows():
        try:
            esquema.validate(fila.to_frame().T)
            valids.append(fila)
        except pa.errors.SchemaError as e:
            fila_error = fila.copy()
            fila_error["motiu_error"] = str(e)
            fila_error["timestamp_error"] = pd.Timestamp.now()
            invalids.append(fila_error)

    df_valids = pd.DataFrame(valids)
    df_invalids = pd.DataFrame(invalids)

    if not df_invalids.empty:
        df_invalids.to_sql(taula_quarantena, conn, if_exists="append", index=False)
        import logging
        logging.warning(
            "Registres enviats a quarantena: %d de %d (%.1f%%)",
            len(df_invalids), len(df), len(df_invalids) / len(df) * 100,
        )

    return df_valids

---

dbt: transformacions SQL versionades

dbt (data build tool) és l'eina estàndard per gestionar les transformacions SQL dins un Data Warehouse (capa ELT). Permet:

• Escriure transformacions com a fitxers .sql versionats a Git.
• Definir tests automàtics sobre les dades transformades.
• Generar documentació i un gràfic de lineage automàticament.
• Executar les transformacions en ordre respectant les dependències.

Model dbt simple

Un model dbt és simplement un fitxer SQL que retorna un SELECT:

-- models/marts/vendes_resum_diari.sql
{{ config(materialized='table') }}

WITH vendes_netes AS (
    SELECT
        id_venda,
        data_comanda::DATE           AS data,
        id_client,
        import,
        pais
    FROM {{ ref('stg_vendes') }}     -- referencia al model staging
    WHERE import > 0
      AND id_client IS NOT NULL
),

resum AS (
    SELECT
        data,
        pais,
        COUNT(*)                     AS n_vendes,
        SUM(import)                  AS import_total,
        AVG(import)                  AS import_mitja,
        COUNT(DISTINCT id_client)    AS clients_actius
    FROM vendes_netes
    GROUP BY data, pais
)

SELECT * FROM resum

Tests dbt

dbt permet definir tests declaratius en fitxers YAML que s'executen després de construir els models:

# models/marts/schema.yml
version: 2

models:
  - name: vendes_resum_diari
    description: "Resum diari de vendes per país"
    columns:
      - name: data
        description: "Data de la venda"
        tests:
          - not_null
      - name: pais
        description: "Codi de país ISO"
        tests:
          - not_null
          - accepted_values:
              values: ["ES", "FR", "DE", "IT", "PT"]
      - name: n_vendes
        description: "Nombre de vendes del dia"
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"
      - name: import_total
        tests:
          - not_null

# Executar tots els models i tests
dbt run
dbt test

# Generar i obrir la documentació
dbt docs generate
dbt docs serve

---

AC5074/05/03 — Miniactivitat

Se't proporciona el fitxer vendes_errors.csv amb el contingut següent (amb errors deliberats):

id_venda,data,id_client,import,email,pais
1,2026-06-01,101,250.00,client@empresa.cat,ES
2,2026-06-01,102,-50.00,no-es-un-email,XX
3,2026-06-02,,300.00,altre@test.com,FR
1,2026-06-02,104,180.00,valid@correu.es,ES
5,2026-06-03,105,99999999.00,enorme@test.com,DE
6,2026-06-03,106,120.00,,IT

Escriu un script Python (validacio_nom_cognom.py) que:

1. Llegeixi el CSV amb pandas.
2. Defineixi un esquema Pandera amb les restriccions apropiades per a cada columna (tipus, nullabilitat, rangs, format d'email, valors acceptats per a pais).
3. Executi la validació amb lazy=True.
4. Generi un informe de text (o CSV) que llisti cada error trobat: columna, índex de la fila, valor invàlid i regla violada.
5. Separi el DataFrame en registres vàlids i invàlids i mostri el recompte de cada grup.

Inclou comentaris que expliquin per qué cada restricció és necessària.

---

Mòdul M5074 Sistemes de Big Data | Institut Sa Palomera (Blanes) | Curs CEIABD 2026-2027