Declarativo

En lugar de iniciar manualmente varios `writeStream`, cada función devuelve un DataFrame que define un dataset. Las lecturas entre datasets establecen dependencias y el pipeline determina el orden. Esto reduce código operativo, pero no elimina decisiones sobre contrato, incrementabilidad, calidad o coste.

El grafo debe expresar transformaciones de datos, no pasos imperativos con efectos externos. Crear archivos, llamar APIs o mutar tablas arbitrariamente dentro de una función declarativa rompe reevaluación y dificulta optimización. Esos efectos pertenecen a tareas de Jobs alrededor del pipeline.

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

@dp.table(name="orders_bronze")
def orders_bronze():
    return spark.readStream.table("main.raw.orders")

@dp.materialized_view(name="daily_order_totals")
def daily_order_totals():
    return (
        spark.read.table("orders_bronze")
          .groupBy(F.to_date("event_ts").alias("order_date"))
          .agg(F.sum("amount").alias("revenue"))
    )

Una definición con `spark.readStream` produce un flujo continuo o por triggers dentro del pipeline. La tabla persiste resultados y el servicio gestiona checkpoints internos. Es adecuada cuando cada registro nuevo puede procesarse incrementalmente sin recalcular todo el resultado.

No debe leerse una fuente cambiante con `spark.read` y esperar semántica streaming. Tampoco se usa una streaming table para una consulta que necesita revisar cambios arbitrarios en ambos lados de un join batch; una materialized view puede permitir actualización incremental gestionada.

from pyspark import pipelines as dp

@dp.table(
    name="orders_bronze",
    comment="Pedidos crudos con metadatos de origen",
    table_properties={"quality": "bronze"},
)
def orders_bronze():
    return (
        spark.readStream.format("cloudFiles")
          .option("cloudFiles.format", "json")
          .option("cloudFiles.schemaLocation", "/Volumes/main/ops/schemas/orders")
          .load("/Volumes/main/landing/orders")
    )

A diferencia de una vista lógica, la salida se materializa para servir consultas rápidas. La definición suele usar `spark.read.table` porque describe el resultado completo correcto. El motor decide si puede aplicar cambios incrementales o si necesita recomputar según consulta y origen.

Es apropiada para joins, agregaciones y modelos gold donde el resultado puede cambiar por actualizaciones upstream. No promete que toda consulta sea siempre incremental; diseño de claves, filtros y operaciones influye en el plan de refresh y debe observarse en el event log.

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

@dp.materialized_view(name="customer_order_metrics")
def customer_order_metrics():
    orders = spark.read.table("orders_silver")
    return (
        orders.groupBy("customer_id")
          .agg(
              F.countDistinct("order_id").alias("orders"),
              F.sum("amount").alias("lifetime_value"),
              F.max("event_ts").alias("last_order_at"),
          )
    )

Un default flow acompaña la definición normal de un dataset. Flows adicionales pueden unir feeds regionales en una misma streaming table sin construir un `union` monolítico. Un append flow añade filas; AUTO CDC aplica cambios ordenados; `ONCE` ejecuta una carga batch una sola vez salvo full refresh.

Cada flow debe tener identidad y semántica compatibles con el target. Un target de AUTO CDC solo recibe flows AUTO CDC. Para backfill histórico, un append flow `ONCE` aislado puede coexistir con la ingesta continua, siempre que no duplique rangos ya procesados.

CREATE OR REFRESH STREAMING TABLE main.bronze.orders_all;

CREATE FLOW orders_eu AS INSERT INTO main.bronze.orders_all
BY NAME SELECT *, 'eu' AS region
FROM STREAM(main.raw.orders_eu);

CREATE FLOW orders_us AS INSERT INTO main.bronze.orders_all
BY NAME SELECT *, 'us' AS region
FROM STREAM(main.raw.orders_us);

Los archivos Python pueden agruparse por bronze, silver y gold o por dominio, siempre que los nombres de dataset sean únicos. Configuración como catálogo fuente, paths y umbrales entra mediante parámetros del pipeline, no constantes replicadas. Las funciones de transformación puras se prueban fuera de los decoradores.

Dev, test y prod ejecutan el mismo código con targets y permisos distintos. Una actualización se valida en un catálogo aislado y con datos representativos antes de promover. El owner revisa event log, lineage y cambios de esquema tras desplegar.

source_catalog = spark.conf.get("pipelines.source_catalog")
target_catalog = spark.conf.get("pipelines.target_catalog")
lateness = spark.conf.get("pipelines.orders_lateness", "15 minutes")

source_table = f"{source_catalog}.bronze.orders"
target_prefix = f"{target_catalog}.commerce"

assert source_catalog != target_catalog or target_catalog.endswith("_dev")