CDC

Al habilitar `delta.enableChangeDataFeed`, las versiones futuras pueden leerse con `readChangeFeed=true`. La salida incluye `_change_type`, `_commit_version` y `_commit_timestamp`; los updates generan imágenes previa y posterior. CDF no reconstruye cambios anteriores a su activación.

El historial disponible depende de la retención de la tabla y de `VACUUM`. Un consumidor que permanezca caído más allá del horizonte puede no recuperar versiones antiguas. Por eso el SLA de recuperación debe relacionar retención, máxima indisponibilidad y una alternativa de snapshot.

spark.sql("""
ALTER TABLE main.bronze.customers
SET TBLPROPERTIES (delta.enableChangeDataFeed = true)
""")

changes = (
    spark.readStream
      .option("readChangeFeed", "true")
      .table("main.bronze.customers")
      .where("_change_type IN ('insert', 'update_postimage', 'delete')")
)

La hora de llegada no es una secuencia fiable: un update antiguo puede llegar después de uno nuevo. `SEQUENCE BY` debe usar un LSN, número de versión u otra columna monotónica del origen. Si hay empates, un `struct` con campos de desempate produce un orden lexicográfico estable.

El pipeline también debe decidir si un `NULL` borra el valor o significa 'campo no enviado', y cómo identificar borrados. La clave no debe cambiar silenciosamente; una modificación de clave suele modelarse como delete de la antigua e insert de la nueva.

SELECT
  customer_id,
  operation,
  sequence_number,
  event_ts,
  named_struct(
    'sequence_number', sequence_number,
    'event_ts', event_ts
  ) AS cdc_sequence
FROM STREAM(main.bronze.customer_cdc);

En SQL se declara una streaming table target y un flow `AUTO CDC INTO`. En Python se usa la API equivalente de Spark Declarative Pipelines en Lakeflow. La fuente debe ser streaming y la clave, secuencia y reglas de borrado quedan en la definición declarativa.

`APPLY CHANGES` sigue disponible por compatibilidad y comparte sintaxis, pero Databricks recomienda AUTO CDC. En documentación, código nuevo y respuestas de diseño debe aparecer el nombre actual; mencionar APPLY CHANGES solo aclara una configuración anterior o una migración.

CREATE OR REFRESH STREAMING TABLE main.silver.customers_current;

CREATE FLOW customers_cdc_flow AS AUTO CDC INTO
  main.silver.customers_current
FROM STREAM(main.bronze.customer_cdc)
KEYS (customer_id)
APPLY AS DELETE WHEN operation = 'DELETE'
SEQUENCE BY struct(sequence_number, event_ts)
COLUMNS * EXCEPT (operation, sequence_number)
STORED AS SCD TYPE 1;

Tipo 1 sobrescribe atributos y es apropiado para correcciones donde la historia no aporta valor. Tipo 2 conserva versiones con columnas de inicio y fin administradas por el pipeline. Consultas de hechos históricos pueden unir el event time del hecho con el intervalo de la dimensión.

Guardar historia multiplica filas y exige decidir qué columnas disparan una nueva versión. `TRACK HISTORY ON` puede limitar ese conjunto en AUTO CDC. Campos operativos como `ingested_at` no deberían crear una versión de negocio cada vez que cambia.

CREATE OR REFRESH STREAMING TABLE main.silver.customers_history;

CREATE FLOW customers_history_flow AS AUTO CDC INTO
  main.silver.customers_history
FROM STREAM(main.bronze.customer_cdc)
KEYS (customer_id)
APPLY AS DELETE WHEN operation = 'DELETE'
SEQUENCE BY sequence_number
COLUMNS * EXCEPT (operation)
STORED AS SCD TYPE 2
TRACK HISTORY ON (name, segment, country);

Algunas bases entregan una extracción completa diaria. Comparar snapshots permite inferir inserts, updates y deletes, y las APIs AUTO CDC FROM SNAPSHOT automatizan esa materialización. Cada snapshot debe tener una versión estrictamente creciente y representar el conjunto completo esperado.

Una extracción parcial confundida con snapshot completo produciría borrados masivos. Antes de publicar se validan conteos, particiones y marcadores de finalización. Si el origen sí ofrece CDC continuo, AUTO CDC normal evita comparar toda la dimensión y reduce latencia.

snapshot_manifest = {
    "snapshot_version": 2026072101,
    "source_table": "crm.customers",
    "expected_rows": 12_450_230,
    "completed": True,
    "landing_path": "/Volumes/main/landing/customers/2026-07-21/",
}

assert snapshot_manifest["completed"]
assert snapshot_manifest["expected_rows"] > 0