CDC vs Delta CDF y recálculo incremental de KPIs históricos

Objetivo: entender cómo detectar cambios en tablas Delta y actualizar KPIs diarios, semanales o mensuales cuando se corrigen tickets antiguos, sin recalcular innecesariamente todo el histórico.

Esta lección conecta directamente con una decisión de arquitectura: ¿debemos implementar nosotros el procesamiento incremental o dejar que Lakeflow lo optimice?

1. CDC y CDF no son lo mismo

La relación es:

Un detalle actual: Databricks distingue entre Automatic CDF, que calcula cambios mediante metadatos de seguimiento de filas, y el CDF tradicional o legacy, que materializa información de cambios durante las escrituras. Databricks on Google Cloud

2. Ejemplo: un ticket de agosto corregido en noviembre

Tenemos esta tabla Silver:

El KPI de agosto es 150 €.

Pero el 12 de noviembre alguien corrige T001:

UPDATE silver_tickets
SET amount = 80
WHERE ticket_id = 'T001';

Ahora agosto debería tener 130 €, aunque la modificación se haya realizado en noviembre.

¿Cómo sabe nuestro proceso Gold que ha cambiado agosto?

Ahí entra Delta CDF.

3. Activar y consultar Delta CDF

Para un ejemplo compatible con el mecanismo tradicional de Delta CDF, habilitamos el seguimiento de cambios antes de realizar el UPDATE:

ALTER TABLE silver_tickets
SET TBLPROPERTIES (
  delta.enableChangeDataFeed = true
);

Después consultamos los cambios usando versiones Delta:

SELECT
    ticket_id,
    amount,
    event_time,
    _change_type,
    _commit_version
FROM table_changes('silver_tickets', 5, 6);

Los números 5 y 6 son ilustrativos. Debes utilizar versiones que existan y en las que CDF estuviera habilitado.

Una actualización genera dos imágenes:

También existen insert y delete.

La información de CDF incluye _commit_version y _commit_timestamp, que sirven para identificar cuándo ocurrió el cambio en Delta. Databricks on AWS

Importante: en las versiones recientes de Databricks que admiten Automatic CDF, no es necesario habilitar la propiedad legacy. El ejemplo anterior es útil para comprender el mecanismo y trabajar en entornos que todavía utilizan CDF tradicional.

4. Simulador: recalcular agosto sin tocar septiembre

El procesamiento incremental puede actualizar una agregación histórica utilizando las diferencias entre el estado anterior y el nuevo.

Pero hay dos estrategias para implementarlo.

5. Estrategia A: Materialized View con incremental refresh

Esta es la opción más sencilla en Lakeflow.

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

@dp.materialized_view(name="gold_monthly_sales")
def gold_monthly_sales():

    return (
        spark.read.table("silver_tickets")
        .groupBy(
            "pharmacy_id",
            F.date_trunc("month", "event_time")
             .alias("month")
        )
        .agg(
            F.sum("amount").alias("sales")
        )
    )

Cuando cambie un ticket de agosto:

1. Lakeflow detecta que hay cambios en Silver.

2. Analiza si puede actualizar incrementalmente la MV.

3. Si es posible y rentable, actualiza los resultados afectados.

4. Si no puede, realiza un Full Recompute para preservar la corrección.

¿Necesito activar CDF explícitamente?

No necesariamente. En Databricks reciente, con Automatic CDF y row tracking, no tienes que habilitar delta.enableChangeDataFeed para obtener incremental refresh.

Para favorecerlo, Databricks recomienda propiedades como:

ALTER TABLE silver_tickets
SET TBLPROPERTIES (
  delta.enableRowTracking = true,
  delta.enableDeletionVectors = true
);

Estas propiedades requieren compatibilidad con las características Delta correspondientes. No garantizan por sí solas que toda consulta sea incremental. Databricks on Google Cloud

Y recuerda que el incremental refresh de las Materialized Views requiere Serverless.

Esta sería mi primera opción para Gold, siempre que comprobemos con el Event Log que la actualización es realmente incremental.

6. Estrategia B: procesar CDF manualmente

Imagina que quieres controlar tú mismo qué KPIs se actualizan.

Puedes leer los cambios desde la última versión Delta procesada:

changes = (
    spark.read
    .option("readChangeFeed", "true")
    .option("startingVersion", 25)
    .option("endingVersion", 30)
    .table("silver_tickets")
)

Suponemos que la versión 24 fue la última procesada correctamente.

Con este DataFrame podemos calcular diferencias:

from pyspark.sql import functions as F

deltas = (
    changes
    .withColumn(
        "delta_amount",
        F.when(
            F.col("_change_type").isin(
                "delete", "update_preimage"
            ),
            -F.col("amount")
        ).otherwise(F.col("amount"))
    )
    .withColumn(
        "month",
        F.date_trunc("month", "event_time")
    )
    .groupBy("pharmacy_id", "month")
    .agg(
        F.sum("delta_amount").alias("sales_delta")
    )
)

Esto transforma los cambios en movimientos contables:

El resultado de nuestro ejemplo es −20 € para agosto.

Después podemos aplicar los deltas a Gold mediante un MERGE.

MERGE INTO gold_monthly_sales AS target
USING monthly_deltas AS source
ON target.pharmacy_id = source.pharmacy_id
AND target.month = source.month

WHEN MATCHED THEN UPDATE SET
  target.sales = target.sales + source.sales_delta

WHEN NOT MATCHED THEN INSERT (
  pharmacy_id,
  month,
  sales
)
VALUES (
  source.pharmacy_id,
  source.month,
  source.sales_delta
);

Aquí monthly_deltas representa el resultado agregado de los cambios y debe estar disponible como vista o tabla.

Este es un patrón simplificado para una métrica aditiva como SUM. En producción también necesitamos asegurar que cada rango de versiones se aplique una sola vez, gestionar reintentos y persistir el progreso de forma coherente con la actualización de Gold. Si la última fila de un grupo desaparece, también habrá que ajustar o eliminar ese agregado.

Un patrón alternativo, muchas veces más sencillo y robusto, consiste en usar CDF para identificar meses afectados y recalcular esos meses completos desde Silver, sustituyendo sus resultados en Gold de manera atómica.

Ese segundo patrón es especialmente útil si tenemos KPIs como COUNT DISTINCT, medianas o métricas que no se pueden actualizar mediante una suma de diferencias.

7. ¿Qué pasa con los KPIs semanales y mensuales?

Supongamos que T001 pertenecía al 10 de agosto.

Si corregimos su importe, debemos actualizar:

• Las ventas diarias del 10 de agosto.

• Las ventas de la semana correspondiente.

• Las ventas mensuales de agosto.

Si además cambiamos la fecha del ticket del 10 de agosto al 3 de septiembre, debemos actualizar tanto las ventanas originales como las nuevas.

Por eso, al usar CDF, son importantes ambas imágenes de un UPDATE: update_preimage y update_postimage.

8. ¿Cuál elegiría en una Data Platform?

Mi recomendación sería empezar con Materialized Views incrementales para Gold, medir su comportamiento real y recurrir a CDF manual cuando necesitemos garantías específicas sobre qué períodos se recalculan o un control más preciso de costes.

Un último detalle: CDF no es un histórico permanente. Los cambios solo pueden consultarse mientras las versiones necesarias sigan retenidas. Si necesitas auditoría permanente o replays de largo plazo, archiva esos cambios. Databricks on AWS

9. Mini ejercicio

Qué debes recordar:

• CDC es el patrón general; CDF permite obtener cambios entre versiones de tablas Delta.

• AUTO CDC aplica eventos de cambio para mantener entidades.

• Una Materialized View puede aprovechar mecanismos internos de seguimiento de cambios para actualizar resultados incrementalmente, sin que tengamos que leer CDF explícitamente.

• Para KPIs históricos, debemos actualizar los períodos de negocio afectados por las versiones anteriores y nuevas de los registros.

En la última lección compararemos dbt y Lakeflow y veremos cuándo conviene combinar ambas tecnologías en lugar de elegir solamente una.