Integritat i Qualitat de Dades en Sistemes Distribuïts

Introducció

El 2018, IBM estimava que la mala qualitat de les dades costava als EUA 3,1 bilions de dòlars anuals. En un món on les empreses prenen decisions crítiques basades en dades —d'un pla de producció a un model de risc creditici—, la qualitat de les dades deixa de ser un tema tècnic per convertir-se en una qüestió estratègica de negoci.

En sistemes distribuïts, la qualitat de les dades és especialment complexa: les dades arriben de desenes de fonts heterogènies (ERP, CRM, IoT, xarxes socials), passen per múltiples transformacions i s'emmagatzemen en sistemes amb consistència eventual. Sense eines específiques de validació i governança, un Data Lake es degrada ràpidament en un "data swamp" on ningú confia en les dades.

---

1. Les sis dimensions de la qualitat de dades

1.1 Framework estàndard

El model internacional ISO 8000 i el DAMA (Data Management Association) defineixen sis dimensions de qualitat de dades que tot professional ha de conèixer:

Dimensió	Definició	Exemple de problema	Mètrica típica
Completeness	Totes les dades necessàries existeixen	Camp codi_postal nul al 30% de clients	% files sense nuls
Consistency	Les dades son coherents entre sistemes	El CRM diu que el client és de Barcelona, el ERP de Madrid	% discrepàncies entre fonts
Accuracy	Les dades reflecteixen la realitat	Una venda registrada per 0€ per error d'entrada	% valors vàlids en rangs esperats
Timeliness	Les dades estan actualitzades quan cal	Un inventari que s'actualitza una vegada al dia en un e-commerce d'alta freqüència	Retard màxim (freshness)
Uniqueness	No hi ha duplicats	Un client registrat tres vegades amb el mateix email	% duplicats
Validity	Les dades compleixen el format i les regles de negoci	Un NIF invàlid, un email sense "@", una data de naixement en el futur	% valors que passen les regles

xychart-beta
    title "Auditoria de Qualitat - Dataset Vendes 2024"
    x-axis ["Completeness", "Consistency", "Accuracy", "Timeliness", "Uniqueness", "Validity"]
    y-axis "Puntuació (0-100)" 0 --> 100
    bar [95, 72, 88, 61, 99, 84]

1.2 Priorització i impacte de negoci

No totes les dimensions son iguals de crítiques per a tots els casos d'ús. Una recomanació pràctica:

• Model de frau bancari: Accuracy i Timeliness son crítiques (un fals negatiu pot costar milions)
• Informe de vendes mensual: Completeness i Consistency son les prioritàries
• Sistema de recomanació: Uniqueness és crítica (un client duplicat genera recomanacions incoherents)

---

2. Integritat a HDFS: checksums i fsck

2.1 CRC32C i la validació de blocs

HDFS usa checksums CRC32C per detectar corrupció de dades a nivell de bloc. Quan un DataNode llegeix un bloc, recalcula el checksum i el compara amb el valor emmagatzemat. Si hi ha discrepància, el bloc es marca com a corrupte i HDFS usa una de les rèpliques sanes per recuperar-lo.

# Verificar la integritat de tots els fitxers d'un directori
hdfs fsck /data/raw/ -files -blocks -locations

# Exemple de sortida:
# /data/raw/vendes_2024_01.parquet 52428800 bytes, replicated: replication=3,
#   1 block(s): OK
# Status: HEALTHY
# Total size: 52428800 B
# Total dirs: 1
# Total files: 1
# Total symlinks: 0
# Total blocks (validated): 1 (avg. block size 52428800 B)
# Minimally replicated blocks: 1 (100.0 %)
# Over-replicated blocks: 0 (0.0 %)
# Under-replicated blocks: 0 (0.0 %)
# Mis-replicated blocks: 0 (0.0 %)
# Default replication factor: 3
# Average block replication: 3.0
# Missing blocks: 0
# Corrupt blocks: 0

# Verificar i reparar blocs sous-replicats
hdfs fsck / -includeSnapshots -list-corruptfileblocks

# Forçar la re-replicació d'un fitxer
hdfs dfsadmin -setReplication 3 /data/raw/fitxer_important.parquet

2.2 Simulació de corrupció i recuperació

En un entorn de test, pots simular la corrupció d'un bloc per veure com HDFS reacciona:

# 1. Identifica la ubicació física del bloc
hdfs fsck /data/raw/test.parquet -files -blocks -locations
# Sortida: /data/hadoop/dfs/data/current/BP-xxxx/current/finalized/
#          subdir0/subdir0/blk_1073741825

# 2. Corromp el bloc (en un DataNode, per a proves!)
# dd if=/dev/urandom of=/ruta/al/bloc bs=1024 count=1 conv=notrunc

# 3. HDFS detecta la corrupció automàticament en el proper heart beat
# i activa la recuperació des d'una rèplica sana

# 4. Verifica que s'ha recuperat
hdfs fsck /data/raw/test.parquet

---

3. Great Expectations

3.1 Conceptes fonamentals

Great Expectations (GE) és el framework estàndard per a validació de dades en pipelines Python. La seva filosofia és convertir les expectatives sobre les dades en codi testejable i documentació automàtica.

Conceptes clau:

• Expectation: una afirmació sobre les dades. Ex: "la columna email no ha de tenir nuls".
• Expectation Suite: un conjunt d'expectations per a un dataset concret.
• Checkpoint: un punt de validació que aplica una suite i desa el resultat.
• Data Docs: documentació HTML auto-generada amb l'historial de validacions.

3.2 Instal·lació i exemple complet

# Instal·lació
# pip install great_expectations

import great_expectations as gx
import pandas as pd

# Crear o carregar el context de GE
context = gx.get_context()

# Carregar el dataset a validar
df = pd.read_csv("/data/vendes_2024_01.csv")

# Crear un Data Source per a Pandas
datasource = context.sources.add_pandas("vendes_source")
asset = datasource.add_dataframe_asset("vendes_2024_01")
batch_request = asset.build_batch_request(dataframe=df)

# Crear una Expectation Suite
suite = context.add_or_update_expectation_suite("suite_vendes")
validator = context.get_validator(batch_request=batch_request,
                                   expectation_suite=suite)

# --- Definir les expectations ---

# Completeness: columnes obligatòries sense nuls
validator.expect_column_values_to_not_be_null("id_venda")
validator.expect_column_values_to_not_be_null("data")
validator.expect_column_values_to_not_be_null("producte")
validator.expect_column_values_to_not_be_null("preu")

# Uniqueness: id_venda ha de ser únic
validator.expect_column_values_to_be_unique("id_venda")

# Validity: preu ha d'estar en un rang raonable
validator.expect_column_values_to_be_between(
    "preu", min_value=0.01, max_value=10000.00
)

# Validity: quantitat ha de ser un enter positiu
validator.expect_column_values_to_be_between(
    "quantitat", min_value=1, max_value=1000
)

# Validity: format de data correcte (ISO 8601)
validator.expect_column_values_to_match_regex(
    "data",
    r"^\d{4}-\d{2}-\d{2}$"
)

# Validity: categoria en llista de valors acceptats
validator.expect_column_values_to_be_in_set(
    "categoria",
    ["Informatica", "Periferics", "Xarxes", "Audio", "Video"]
)

# Accuracy: preu no pot ser exactament 0 (seria un error d'entrada)
validator.expect_column_values_to_not_be_null("preu")

# Timeliness: data no pot ser anterior a l'1 de gener de 2020
validator.expect_column_values_to_be_between(
    "data",
    min_value="2020-01-01",
    max_value="2030-12-31",
    parse_strings_as_datetimes=True
)

# Consistència estadística: la mitjana del preu ha d'estar en rang esperat
validator.expect_column_mean_to_be_between(
    "preu", min_value=50.0, max_value=1000.0
)

# Guardar les expectations
validator.save_expectation_suite(discard_failed_expectations=False)

print("Expectation Suite creada i desada.")

3.3 Executar la validació i generar Data Docs

# Crear un Checkpoint per a executar les validacions
checkpoint = context.add_or_update_checkpoint(
    name="checkpoint_vendes_diari",
    validations=[{
        "batch_request": batch_request,
        "expectation_suite_name": "suite_vendes",
    }]
)

# Executar el Checkpoint
resultats = checkpoint.run()

# Mostrar resum dels resultats
print(f"\n--- RESULTATS DE VALIDACIÓ ---")
print(f"Totes les expectations complides: {resultats.success}")
print(f"\nDetall per expectation:")
for result in resultats.run_results.values():
    for expectation_result in result["validation_result"]["results"]:
        estat = "OK" if expectation_result["success"] else "FALLA"
        expectation_type = expectation_result["expectation_config"]["expectation_type"]
        col = expectation_result["expectation_config"]["kwargs"].get("column", "global")
        print(f"  [{estat}] {col}: {expectation_type}")

# Generar Data Docs (documentació HTML)
context.build_data_docs()
context.open_data_docs()
print("\nData Docs generades i obertes al navegador.")

3.4 Integració amb Apache Airflow

# DAG d'Airflow amb Great Expectations
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import great_expectations as gx

default_args = {
    'owner': 'data-team',
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
    'email_on_failure': True,
    'email': ['data-team@empresa.cat'],
}

def validar_dades_vendes(**kwargs):
    context = gx.get_context()
    checkpoint = context.get_checkpoint("checkpoint_vendes_diari")
    resultats = checkpoint.run()

    if not resultats.success:
        raise ValueError("Validació de dades fallada! Revisa els Data Docs.")

    return "Validació completada amb exit"

with DAG(
    dag_id="pipeline_vendes_amb_validacio",
    default_args=default_args,
    start_date=datetime(2024, 1, 1),
    schedule_interval="@daily",
    catchup=False,
) as dag:

    t_validar = PythonOperator(
        task_id="validar_dades",
        python_callable=validar_dades_vendes,
    )
    # ... altres tasques del pipeline

Miniactivitat

Descarrega el dataset de NYC Taxi (un mes) i crea una Expectation Suite amb almenys 8 expectations:

• 3 de completeness (columnes sense nuls)
• 2 de validity (rangs numèrics)
• 1 d'uniqueness (clau primària)
• 2 de consistència estadística (mitjana, desviació)

Executa la validació i explica quines expectations fallen i per qué.

---

4. dbt: tests de dades en pipelines SQL

4.1 Qué és dbt i per qué canvia les regles del joc

dbt (data build tool) és l'eina que ha redefinit el rol de l'Analytics Engineer els darrers anys. Permet escriure transformacions SQL com a models version-controlats, amb tests automàtics i documentació integrada. La filosofia dbt és "Transform data in your warehouse": les dades s'ingestionen "brutes" (ELT), i dbt les transforma dins del data warehouse o lakehouse.

Raw Data (bronze)  →  Staging Models (silver)  →  Mart Models (gold)
Debezium → S3     →  dbt stg_vendes           →  dbt marts_vendes_mensual

4.2 Tests de schema (built-in)

# models/staging/schema.yml
version: 2

models:
  - name: stg_vendes
    description: "Vendes netes i tipificades des de la font bruta"
    columns:
      - name: id_venda
        description: "Identificador únic de la venda"
        tests:
          - not_null
          - unique

      - name: preu_unitari
        description: "Preu unitari del producte en euros"
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 10000

      - name: categoria
        description: "Categoria del producte"
        tests:
          - not_null
          - accepted_values:
              values: ['Informatica', 'Periferics', 'Xarxes', 'Audio', 'Video']

      - name: data_venda
        tests:
          - not_null

  - name: marts_vendes_mensual
    description: "Agregació mensual de vendes per categoria"
    columns:
      - name: any_mes
        tests:
          - not_null
          - unique

      - name: total_ingressos
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0

4.3 Tests personalitzats amb SQL

-- tests/test_preu_major_que_cost.sql
-- Test de negoci: el preu de venda ha de ser sempre major que el cost
SELECT
    v.id_venda,
    v.preu_unitari,
    p.cost_unitari
FROM {{ ref('stg_vendes') }} v
JOIN {{ ref('stg_productes') }} p ON v.id_producte = p.id_producte
WHERE v.preu_unitari <= p.cost_unitari

-- tests/test_vendes_sense_client.sql
-- Test d'integritat referencial: no han d'existir vendes sense client
SELECT v.id_venda
FROM {{ ref('stg_vendes') }} v
LEFT JOIN {{ ref('stg_clients') }} c ON v.id_client = c.id_client
WHERE c.id_client IS NULL

4.4 Sources i freshness checks

# models/sources.yml
version: 2

sources:
  - name: raw_vendes
    description: "Dades en brut provinents de la BBDD de producció"
    database: datalake
    schema: raw
    tables:
      - name: vendes
        description: "Taula de vendes en brut"
        loaded_at_field: _loaded_at
        freshness:
          warn_after: {count: 12, period: hour}
          error_after: {count: 24, period: hour}
        columns:
          - name: id_venda
            tests:
              - not_null

# Executar tots els tests
dbt test

# Executar tests d'una sola taula
dbt test --select stg_vendes

# Verificar freshness de les fonts
dbt source freshness

# Generar documentació
dbt docs generate
dbt docs serve  # Obre la documentació al navegador

4.5 Exemple de model dbt complet

-- models/staging/stg_vendes.sql
WITH font_bruta AS (
    SELECT * FROM {{ source('raw_vendes', 'vendes') }}
),

neta AS (
    SELECT
        id_venda,
        CAST(data AS DATE)                           AS data_venda,
        TRIM(LOWER(producte))                        AS producte,
        CAST(preu AS NUMERIC(10, 2))                 AS preu_unitari,
        CAST(quantitat AS INTEGER)                   AS quantitat,
        CAST(quantitat * preu AS NUMERIC(12, 2))     AS import_brut,
        COALESCE(descompte, 0.0)                     AS descompte,
        CAST(quantitat * preu * (1 - COALESCE(descompte, 0))
             AS NUMERIC(12, 2))                      AS import_net,
        id_client,
        id_venedor,
        CURRENT_TIMESTAMP                            AS _processat_a
    FROM font_bruta
    WHERE
        preu > 0
        AND quantitat > 0
        AND data IS NOT NULL
        AND id_venda IS NOT NULL
)

SELECT * FROM neta

---

5. Gestió de metadades

5.1 Apache Atlas: lineage i classificació

Apache Atlas és el sistema de governança de metadades d'Hadoop. Permet:

• Data Lineage: traçar el camí d'una dada des del seu origen fins al dashboard final
• Classificació: etiquetar camps com a PII, confidencial, restringit
• Catàleg: descobrir quines taules i columnes existeixen i qué signifiquen

PostgreSQL.vendes.preu_unitari
    → HDFS.raw.vendes (Kafka Connect, cada 10 min)
    → Delta Lake.staging.stg_vendes (Spark job, cada hora)
    → Delta Lake.marts.vendes_mensual (dbt, cada nit)
    → Metabase.Dashboard_Vendes.KPI_Ingressos (actualitzat cada nit)

5.2 DataHub (LinkedIn): descobriment de dades

DataHub (ara open source i mantingut per Acryl Data) és el successor espiritual d'Atlas amb una UI molt més moderna. Permet:

• Cerca de taules per paraula clau, propietari o tag
• Vista de lineage interactiva (qui produeix, qui consumeix cada dataset)
• Integració amb dbt (importa automàticament la documentació de dbt)
• Slack bot: respon preguntes sobre el catàleg de dades

# Instal·lació de DataHub en local
docker compose -f docker-compose.quickstart.yml up -d

# Ingestió de metadades de dbt
datahub ingest -c datahub_dbt.yml

# Accedir a la UI
# http://localhost:9002

---

6. Governança de dades

6.1 Data Mesh: propietat descentralitzada

El Data Mesh, proposat per Zhamak Dehghani (2019), és un paradigma que tracta les dades com a productes i distribueix la responsabilitat de cada domini de dades als equips de negoci:

graph TD
    subgraph Vendes["Domini de Vendes"]
        PV[Producte: Dataset Vendes Diaries]
    end

    subgraph Logistica["Domini de Logistica"]
        PL[Producte: Dataset Enviaments]
    end

    subgraph Marketing["Domini de Marketing"]
        PM[Producte: Dataset Campanyes]
    end

    subgraph Plataforma["Plataforma Self-Serve - IT Central"]
        INFRA[Infraestructura de Dades - S3, Kafka, Spark, Airflow]
        QUALITAT[Estandards de Qualitat - Great Expectations, dbt]
        CATALOG[Cataleg de Dades - DataHub]
    end

    PV --> INFRA
    PL --> INFRA
    PM --> INFRA
    INFRA --> QUALITAT
    INFRA --> CATALOG
    QUALITAT --> Consum[Consumers - Data Scientists, BI]
    CATALOG --> Consum

6.2 Data Contracts

Un Data Contract és un acord formal entre el productor i els consumidors d'un dataset. Defineix:

• Esquema: camps, tipus, nullable
• SLA de qualitat: freshness màxima, % de nuls màxim
• SLA de disponibilitat: uptime mínim del pipeline
• Versionat: política de compatibilitat d'esquema

# Exemple de Data Contract (format YAML)
dataContract:
  id: "vendes-diaries-v1"
  version: "1.2.0"
  producer:
    team: "Equip de Vendes"
    owner: "Joan Garcia"
    contact: "jgarcia@empresa.cat"

  schema:
    - name: id_venda
      type: string
      nullable: false
      unique: true
    - name: data
      type: date
      nullable: false
    - name: import_net
      type: decimal(12,2)
      nullable: false
      constraints:
        min: 0.01
        max: 100000.00

  sla:
    freshness: "max 2 hores"
    nullability_threshold: "< 0.1% per a camps obligatoris"
    availability: "99.5% uptime mensual"

  consumers:
    - team: "BI"
      use: "Dashboard de Vendes Diari"
    - team: "ML"
      use: "Model de predicció de demanda"

---

7. RGPD i Big Data

7.1 El repte del dret a l'oblit en un Data Lake

El Reglament General de Protecció de Dades (RGPD) obliga les empreses a eliminar les dades personals d'un individu quan aquest ho sol·licita. En un sistema Big Data distribuït, això és tècnicament complex:

• Les dades d'un client poden estar en HDFS (en múltiples rèpliques), Kafka (log immutable), Delta Lake (múltiples versions per time travel), backups i caches de sistemes de BI.
• La deduplicació i la pseudonimització de dades distribuïdes és un repte de consistència.

7.2 GDPR Delete amb Delta Lake

from delta.tables import DeltaTable
from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("GDPR_Delete") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .getOrCreate()

# Eliminar totes les dades del client C045 (ha exercit el dret a l'oblit)
deltaTable = DeltaTable.forPath(spark, "/data/delta/vendes")

print("Dades del client C045 ABANS de l'eliminació:")
spark.read.format("delta").load("/data/delta/vendes") \
    .filter("id_client = 'C045'") \
    .show()

# Eliminació de les files (operació ACID)
deltaTable.delete("id_client = 'C045'")

print("Dades del client C045 DESPRES de l'eliminació:")
spark.read.format("delta").load("/data/delta/vendes") \
    .filter("id_client = 'C045'") \
    .show()
# → 0 files

# IMPORTANT: el time travel encara permet accedir a les dades anteriors!
# Per a una eliminació RGPD completa, cal executar VACUUM
deltaTable.vacuum(retentionHours=0)  # Eliminació permanent

print("Eliminació RGPD completada. Fitxers antics eliminats.")

VACUUM 0 hores en producció

vacuum(retentionHours=0) elimina permanentment tots els fitxers que no formen part de la versió actual, inclosos els que permetrien el Time Travel. Abans d'executar-lo, has de deshabilitar la comprovació de seguretat de Delta Lake:

spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")

Usa-ho únicament per a peticions RGPD documentades.

7.3 Pseudonimització en streaming

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
import hashlib

spark = SparkSession.builder.appName("Pseudonimitzacio").getOrCreate()

# Funció de pseudonimització (hash SHA-256 amb salt)
SALT = "iabd2025-salt-secret"

@F.udf(returnType="string")
def pseudonimitzar(valor):
    if valor is None:
        return None
    return hashlib.sha256(f"{SALT}:{valor}".encode()).hexdigest()[:16]

# Aplicar a un stream de dades de clients
df_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "clients") \
    .load()

df_pseudonim = df_stream \
    .select(F.from_json(F.col("value").cast("string"), schema_client).alias("data")) \
    .select("data.*") \
    .withColumn("email_hash", pseudonimitzar(F.col("email"))) \
    .withColumn("nom_hash", pseudonimitzar(F.col("nom"))) \
    .drop("email", "nom")  # Eliminar les dades originals del PII

df_pseudonim.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "/checkpoints/clients_pseudonim") \
    .start("/data/delta/clients_pseudonim")

---

8. Deduplicació a escala

8.1 Bloom Filters

Un Bloom Filter és una estructura de dades probabilística que permet respondre "¿ha aparegut mai aquest valor?" amb molt poca memòria. Accepta falsos positius (pot dir "sí" quan la resposta és "no") però mai falsos negatius.

Cas d'ús: en un pipeline que reb milions de clics, usar un Bloom Filter per descartar de forma ràpida els events que ja has processat.

from pyspark.sql import functions as F

# PySpark: Bloom filter per a deduplicació aproximada
df_nous_ids = df_stream.select("id_event") \
    .filter(~F.expr("might_contain(bloom_filter, id_event)"))

8.2 MinHash LSH: deduplicació de text

Locality-Sensitive Hashing (LSH) permet trobar parells de registres "similars" en un gran dataset sense comparar-los tots a tots. És especialment útil per a deduplicació de noms de clients o adreces:

• "Joan Garcia" ≈ "J. Garcia" ≈ "JOAN GARCIA"
• "Carrer Major 5, Barcelona" ≈ "C/ Major, 5 - Barcelona"

from pyspark.ml.feature import MinHashLSH, Tokenizer, HashingTF
from pyspark.sql import functions as F

# Convertir noms en representació numèrica (n-grames)
tokenizer = Tokenizer(inputCol="nom_client", outputCol="tokens")
hashingTF = HashingTF(inputCol="tokens", outputCol="features", numFeatures=1000)
minhash = MinHashLSH(inputCol="features", outputCol="hashes", numHashTables=5)

# Pipeline
df_tokens = tokenizer.transform(df_clients)
df_features = hashingTF.transform(df_tokens)
model = minhash.fit(df_features)
df_transformed = model.transform(df_features)

# Trobar parells similars (Jaccard distance < 0.3)
similars = model.approxSimilarityJoin(
    df_transformed, df_transformed,
    threshold=0.3,
    distCol="distancia_jaccard"
).filter("datasetA.id_client < datasetB.id_client")

print("Parells de clients potencialment duplicats:")
similars.select(
    "datasetA.nom_client",
    "datasetB.nom_client",
    "distancia_jaccard"
).orderBy("distancia_jaccard").show(20)

---

9. Schema Registry i validació de format en temps real

Quan múltiples equips produeixen missatges a Kafka, és comú que un missatge mal format trenqui el pipeline del consumidor. El Schema Registry de Confluent prevé aquest problema:

Productor → [Schema Registry valida l'esquema] → Kafka → Consumidor

from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka import SerializingProducer

# Configuració del Schema Registry
schema_registry_conf = {'url': 'http://schema-registry:8081'}
schema_registry_client = SchemaRegistryClient(schema_registry_conf)

# Esquema Avro de la venda
schema_str = """
{
  "type": "record",
  "name": "Venda",
  "namespace": "cat.sapalomera.iabd",
  "fields": [
    {"name": "id_venda", "type": "string"},
    {"name": "producte", "type": "string"},
    {"name": "preu", "type": {"type": "bytes", "logicalType": "decimal",
                              "precision": 10, "scale": 2}},
    {"name": "quantitat", "type": "int"},
    {"name": "timestamp", "type": {"type": "long", "logicalType": "timestamp-millis"}}
  ]
}
"""

avro_serializer = AvroSerializer(
    schema_registry_client,
    schema_str,
    lambda obj, ctx: obj  # Convertir l'objecte Python a dict
)

producer = SerializingProducer({
    'bootstrap.servers': 'kafka:9092',
    'value.serializer': avro_serializer
})

# Si el productor envia un missatge que no compleix l'esquema,
# el Schema Registry rebutja la publicació
producer.produce(
    topic="vendes",
    value={"id_venda": "V999", "producte": "Test",
           "preu": b'\x00\x01', "quantitat": 1,
           "timestamp": 1705276800000}
)

---

Exercici pràctic final

Implementa un pipeline de qualitat de dades complet per al dataset de NYC Taxi:

1. Audita la qualitat: Usa pandas + Great Expectations per crear una suite amb almenys 10 expectations cobrines les 6 dimensions de qualitat.
2. Crea un model dbt: Instal·la dbt-duckdb, crea un model de staging que netegi el dataset, i afegeix almenys 5 tests de schema.
3. Implementa RGPD delete: Si el dataset tingués camps PII (imagina que VendorID és un identificador de persona), implementa l'eliminació RGPD amb Delta Lake.
4. Genera la documentació: Usa dbt docs generate i Great Expectations Data Docs per generar documentació HTML de la qualitat del dataset.

Documenta cada decisió de qualitat: per qué has definit els rangs que has definit, qué significa cada expectation en termes de negoci.

---

Tema 2 | Mòdul 5075 Big Data Aplicat | Institut Sa Palomera (Blanes) | Curs IABD 2026-2027