Nuestro primer pipeline real en Databricks

Hoy vamos a construir un pipeline de verdad, sin Kafka todavía, para centrarnos en cómo funciona Lakeflow.

Al finalizar tendrás una Streaming Table, una Materialized View y un DAG visible en Databricks.

1. La arquitectura que vamos a construir

2. Preparar los datos de entrada

Abre Databricks y entra en el SQL Editor. Elige un catálogo y esquema de pruebas donde puedas crear tablas.

Por ejemplo, vamos a usar dev_catalog.demo_lakeflow. Sustituye estos nombres por un catálogo y esquema reales de tu workspace.

Ejecuta:

CREATE SCHEMA IF NOT EXISTS
  dev_catalog.demo_lakeflow;

CREATE TABLE IF NOT EXISTS
  dev_catalog.demo_lakeflow.bronze_tickets
AS
SELECT *
FROM VALUES
  (1, 109, 100.0),
  (2, 197,  50.0),
  (3, 109,  75.0),
  (4, 197, -10.0)
AS t(ticket_id, pharmacy_id, amount);

Tenemos cuatro tickets, incluido uno con importe negativo que queremos excluir.

3. Crear el pipeline

En Databricks:

1. Abre New → ETL pipeline.

2. Ponle el nombre learning_lakeflow.

3. Selecciona el mismo catálogo y esquema que utilizaste antes.

4. Elige comenzar con una transformación en Python.

5. Usa el modo Triggered y, si está habilitado, compute Serverless.

La interfaz actual de Databricks permite crear y ejecutar pipelines directamente desde Lakeflow Pipelines Editor.

Databricks on AWS+2

4. Escribir nuestro primer pipeline

En el archivo Python de transformaciones, sustituye el contenido de ejemplo por lo siguiente:

from pyspark import pipelines as dp
from pyspark.sql import functions as F


# SILVER: Streaming Table
@dp.table(name="silver_tickets")
def silver_tickets():
    return (
        spark.readStream.table("bronze_tickets")
        .filter(F.col("amount") > 0)
    )


# GOLD: Materialized View
@dp.materialized_view(name="gold_sales")
def gold_sales():
    return (
        spark.read.table("silver_tickets")
        .groupBy("pharmacy_id")
        .agg(
            F.sum("amount").alias("total_sales")
        )
    )

Fíjate en estas tres cosas:

• @dp.table declara una Streaming Table porque devolvemos un DataFrame de streaming.

• @dp.materialized_view declara nuestra agregación.

• spark.read.table("silver_tickets") establece automáticamente la dependencia entre Gold y Silver.

Aquí no hay instrucciones manuales de escritura, checkpoints ni coordinación de jobs.

5. Ejecutar y verificar

Pulsa Run pipeline.

Una vez que termine, abre el Pipeline Graph para inspeccionar el DAG. Deberías encontrar silver_tickets y gold_sales, con una dependencia desde Silver hasta Gold.

Después, desde SQL Editor, consulta:

SELECT *
FROM dev_catalog.demo_lakeflow.gold_sales
ORDER BY pharmacy_id;

El resultado esperado es:

El ticket de -10.0 no se ha incluido porque Silver lo ha filtrado.

6. Experimento: añade un ticket nuevo

Ahora viene lo interesante. Sin modificar el código del pipeline, ejecuta en SQL Editor:

INSERT INTO dev_catalog.demo_lakeflow.bronze_tickets
VALUES (5, 109, 200.0);

Vuelve al pipeline y pulsa Run pipeline.

Cuando termine, consulta gold_sales otra vez.

¿Por qué?

• Silver ha procesado el nuevo ticket mediante la lectura incremental.

• Gold ha actualizado el agregado.

• No hemos tenido que programar manualmente qué tablas deben ejecutarse ni coordinar sus dependencias.

Este ejemplo presupone que Bronze recibe únicamente inserciones. No hagas UPDATE ni DELETE sobre ella todavía: una lectura streaming convencional no trata esos cambios como nuevas filas y puede fallar. Databricks on AWS

7. Comprueba lo que has aprendido

Idea clave: Lakeflow gestiona las dependencias y la ejecución, mientras tú describes las transformaciones.

Y algo que exploraremos más adelante: Lakeflow gestiona los checkpoints de los flows de streaming, pero eso no elimina la necesidad de entender la semántica de los datos, la idempotencia y la recuperación ante errores.

Próxima lección: sustituiremos la tabla Bronze de prueba por Azure Event Hubs usando el protocolo Kafka. Será nuestro primer paso hacia una arquitectura de producción.