De Bronze a Silver: Avro, parsing y limpieza

Objetivo: aprender a transformar los bytes que recibimos de Kafka en una tabla Silver con datos tipados, limpios y preparados para su consumo.

Hoy veremos tres conceptos:

1. Cómo deserializar Avro.

2. Cómo limpiar y validar los datos.

3. Cuándo utilizar spark.read o spark.readStream.

1. Nuestra arquitectura

Recuerda: Bronze almacena los datos tal como llegan. Silver les da estructura y significado.

2. ¿Qué significa deserializar Avro?

Imagina que Kafka recibe este ticket:

{
  "ticket_id": "T001",
  "pharmacy_id": 109,
  "amount": 125.50
}

Pero en Kafka no tenemos este JSON. Tenemos una secuencia de bytes codificados en Avro.

Para interpretarlos necesitamos conocer su esquema:

{
  "type": "record",
  "name": "Ticket",
  "fields": [
    {"name": "ticket_id", "type": "string"},
    {"name": "pharmacy_id", "type": "int"},
    {"name": "amount", "type": "double"}
  ]
}

El esquema describe cómo interpretar esos bytes: el nombre, el orden y los tipos de los campos.

La función que necesitas conocer es from_avro().

Databricks permite utilizarla tanto en transformaciones batch como streaming. Databricks on AWS

3. Nuestro primer Silver con Lakeflow

Para simplificar, utilizaremos un esquema Avro fijo y supondremos que value contiene directamente un registro Avro binario, sin prefijos adicionales.

import json
from pyspark import pipelines as dp
from pyspark.sql import functions as F
from pyspark.sql.avro.functions import from_avro

# Contrato Avro de ejemplo
AVRO_SCHEMA = json.dumps({
    "type": "record",
    "name": "Ticket",
    "fields": [
        {"name": "ticket_id", "type": "string"},
        {"name": "pharmacy_id", "type": "int"},
        {"name": "amount", "type": "double"}
    ]
})

@dp.table(name="silver_tickets")
def silver_tickets():

    bronze = spark.readStream.table(
        "bronze_tickets"
    )

    parsed = bronze.withColumn(
        "ticket",
        from_avro(
            F.col("value"),
            AVRO_SCHEMA
        )
    )

    return (
        parsed
        .select(
            F.col("ticket.ticket_id").alias("ticket_id"),
            F.col("ticket.pharmacy_id").alias("pharmacy_id"),
            F.col("ticket.amount").alias("amount"),
            F.col("kafka_timestamp")
        )
        .filter(F.col("amount") > 0)
    )

Observa que seguimos sin utilizar .write ni gestionar checkpoints manualmente.

La transformación lee los nuevos registros de Bronze, interpreta los bytes Avro, selecciona los campos y elimina los tickets cuyo importe no sea positivo.

El código supone que bronze_tickets procede de la lección 5 y que todos los mensajes corresponden al esquema mostrado. Para mensajes de producción con múltiples versiones necesitaremos resolver el esquema correcto.

4. Experimento: ¿qué registros llegan a Silver?

En producción es importante distinguir entre un error técnico de deserialización y un dato que incumple una regla de negocio. No siempre queremos descartarlos de la misma forma.

En la lección 7 aprenderemos a controlar esto utilizando Lakeflow Expectations.

5. La diferencia entre read y readStream

Esta es probablemente la parte más importante de la lección.

spark.readStream.table("bronze_tickets")

spark.read.table("bronze_tickets")

Ejemplo fundamental:

Supongamos que un ticket ya procesado cambia de 100 € a 150 €.

• Con readStream, una Streaming Table convencional no propaga automáticamente ese UPDATE y puede fallar al detectar el cambio en Delta.

• Con read, una Materialized View puede reflejar el nuevo valor al refrescarse.

Por eso no deberíamos elegir siempre Streaming Tables únicamente porque sean más eficientes.

Databricks on AWS+1

6. ¿Y cómo lo haríamos con Apicurio?

En vuestra arquitectura hay una diferencia importante: no utilizáis un único esquema Avro fijo.

Tenéis contratos, diferentes versiones y headers para identificar los esquemas.

Una posible arquitectura sería:

Hay un matiz técnico: si Apicurio devuelve mensajes en formato compatible con Confluent Schema Registry, podemos estudiar la deserialización integrada con from_avro y Schema Registry. Si el mensaje lleva Avro puro y la identificación del esquema va exclusivamente en headers, necesitaremos resolverlo de otra manera. Databricks on AWS

Para producción, procuraría evitar consultas HTTP a Apicurio por cada evento. Es preferible resolver las versiones de los esquemas de forma controlada y reutilizarlas durante el procesamiento.

7. Mini ejercicio

Qué debes recordar hoy: from_avro convierte bytes en columnas tipadas; readStream procesa incrementalmente fuentes append-only, mientras que read permite definir una consulta sobre el estado de los datos. La decisión entre Streaming Table y Materialized View depende de cómo cambian los datos, no solamente de su volumen.

En la lección 7 aprenderemos a detectar y controlar datos incorrectos sin depender únicamente de filtros manuales.