Trabajar con tablas Delta Lake en Microsoft Fabric

¿Qué te enseña este módulo?

Este es el módulo más importante de toda la Fase 1. Aquí aprendemos a fondo la tecnología que sostiene absolutamente todo el data engineering en Fabric. Los objetivos oficiales:

1. Entender Delta Lake y las Delta tables en Microsoft Fabric.
2. Crear y gestionar tablas Delta usando Spark.
3. Optimizar tablas Delta.
4. Usar Spark para consultar y transformar datos en Delta tables.
5. Usar Delta tables con Spark Structured Streaming.

Peso en el examen DP-700

Delta Lake aparece transversalmente en todo el examen, pero especialmente en:

Definición

Delta Lake es una storage layer open-source que añade semántica de base de datos relacional al procesamiento data lake basado en Spark.

En cristiano: es el formato que hace que un montón de ficheros Parquet se comporten como una tabla de base de datos con transacciones, schema, versiones, etc. 🌸

Dónde ves que una tabla es Delta

En el Lakehouse explorer de Fabric, las tablas Delta se identifican por el icono triangular delta (Δ) al lado del nombre. Si no ves ese icono, esa "tabla" no es Delta y no tendrá las capacidades avanzadas.

Delta en el ecosistema Fabric

En Fabric, todas las tablas de un lakehouse son Delta tables por defecto. No tienes que hacer nada especial: cuando escribes datos a Tables/, se guardan en formato Delta automáticamente.

Estructura física en OneLake

Cada tabla Delta ocupa una carpeta con esta estructura:

Tables/mi_tabla_delta/
├── part-00000-abc123....snappy.parquet   ← Datos (Parquet)
├── part-00001-def456....snappy.parquet
├── part-00002-ghi789....snappy.parquet
├── ...
└── _delta_log/                            ← Transaction log
    ├── 00000000000000000000.json          ← Cada operación = 1 JSON
    ├── 00000000000000000001.json
    ├── 00000000000000000002.json
    ├── ...
    └── 00000000000000000010.checkpoint.parquet   ← Checkpoint cada N versiones

Los dos ingredientes

A) Ficheros Parquet (datos)

• Formato columnar comprimido.
• Inmutables: nunca se modifican en sitio. Cada cambio genera nuevos ficheros.
• Contienen los valores reales de las filas.

B) Carpeta _delta_log/ (metadatos)

• JSON por cada transacción: describe qué ficheros forman parte de la tabla en cada versión.
• Contiene información de add, remove, schema, metadata.
• Cada N versiones se genera un checkpoint en Parquet (para acelerar la lectura del log).

🎯 Cómo Delta sabe qué ficheros están "activos"

Cuando lees una tabla Delta:

1. Delta lee el último estado consolidado del _delta_log.
2. Ese estado dice: "los ficheros activos son X, Y, Z".
3. Delta lee solo esos ficheros.
4. Los ficheros viejos siguen físicamente ahí (hasta que hagas VACUUM) → habilitan time travel.

¿Cómo se hace UPDATE si Parquet es inmutable?

Delta no modifica en sitio. Cuando haces UPDATE:

1. Delta identifica los ficheros que contienen las filas afectadas.
2. Lee esos ficheros.
3. Aplica los cambios en memoria.
4. Escribe nuevos ficheros con las filas actualizadas.
5. Marca los ficheros viejos como removidos en el nuevo JSON del _delta_log.
6. Nuevas lecturas ven la nueva versión.

Es "copy-on-write". Por eso a veces se acumulan muchos ficheros pequeños → problema del small files, que resolvemos con OPTIMIZE.

Estos son los 5 puntos oficiales de la doc de Microsoft. Apréndetelos porque salen en preguntas del tipo "¿Cuáles son los beneficios de usar Delta?".

🎯 A) Tablas relacionales con operaciones CRUD

Con Spark puedes:

• CREATE tablas Delta
• SELECT filas
• INSERT nuevas filas
• UPDATE filas existentes
• DELETE filas
• MERGE (upsert)

Es decir, como una BBDD relacional, pero encima de tu data lake.

🎯 B) Transacciones ACID

Delta implementa las 4 propiedades ACID sobre Spark:

• Atomicity: una transacción se completa entera o no se aplica nada.
• Consistency: la tabla siempre queda en un estado válido.
• Isolation: procesos concurrentes no interfieren entre sí (serializable isolation).
• Durability: los cambios persisten aunque falle el sistema.

Esto es lo que permite que varios notebooks escriban a la misma tabla simultáneamente sin corromperla.

🎯 C) Versionado de datos y time travel

Como todas las transacciones se registran, puedes:

• Ver el historial completo de cambios.
• Consultar la tabla como estaba en un momento pasado.
• Auditar cambios.
• Recuperarte de errores accidentales.

Ejemplos: versionAsOf, timestampAsOf, DESCRIBE HISTORY.

🎯 D) Soporte para batch Y streaming

Delta puede ser:

• Sink (destino) de un stream → los datos que llegan se acumulan en la tabla.
• Source (fuente) de un stream → cambios en la tabla se procesan como stream.

Con la misma API que para batch. Esto es súper potente para escenarios de IoT, logs, telemetría.

🎯 E) Formato estándar e interoperabilidad

Delta usa Parquet por debajo, que es el formato estándar en data lakes. Esto significa:

• Cualquier motor Parquet puede leerlo.
• El SQL analytics endpoint puede consultar tablas Delta con T-SQL.
• Interoperabilidad con Databricks, Synapse, Snowflake y otros que soporten Delta.

Hay múltiples caminos para crear una tabla Delta. Cada uno tiene su caso de uso.

Método 1: saveAsTable() desde un DataFrame (el más común)

# Cargar datos en un DataFrame
df = spark.read.load('Files/mydata.csv', format='csv', header=True)

# Guardarlos como tabla Delta gestionada
df.write.format("delta").saveAsTable("mytable")

Qué pasa:

• Los datos se guardan como Parquet en Tables/mytable/
• Se crea el _delta_log/
• La tabla aparece en el Lakehouse Explorer bajo Tables
• Es una tabla managed (Fabric gestiona todo)

Método 2: saveAsTable() con path para external table

# Tabla external: metadata en el metastore, datos en Files/
df.write.format("delta").saveAsTable(
    "myexternaltable",
    path="Files/myexternaltable"
)

O apuntando a ADLS externo:

df.write.format("delta").saveAsTable(
    "myexternaltable",
    path="abfss://container@storage.dfs.core.windows.net/myexternaltable"
)

Método 3: DeltaTableBuilder API (crear tabla vacía con schema)

from delta.tables import *

DeltaTable.create(spark) \
  .tableName("products") \
  .addColumn("ProductId", "INT") \
  .addColumn("ProductName", "STRING") \
  .addColumn("Category", "STRING") \
  .addColumn("Price", "FLOAT") \
  .execute()

Uso típico: cuando quieres preparar el schema primero y luego llenar la tabla con datos que vienen por otro camino (streaming, notebooks múltiples, etc.).

Método 4: SQL CREATE TABLE

%%sql
CREATE TABLE salesorders
(
    OrderId INT NOT NULL,
    OrderDate TIMESTAMP NOT NULL,
    CustomerName STRING,
    SalesTotal FLOAT NOT NULL
)
USING DELTA

Sintaxis SQL clásica. Para external:

%%sql
CREATE TABLE MyExternalTable
USING DELTA
LOCATION 'Files/mydata'

Cuando creas una external table con CREATE TABLE ... LOCATION, el schema se infiere de los ficheros Parquet ya presentes en esa ubicación.

Método 5: Escribir Delta sin crear tabla en metastore

A veces quieres guardar datos en formato Delta pero sin registrar la tabla en el metastore:

delta_path = "Files/mydatatable"
df.write.format("delta").save(delta_path)

Escribe ficheros Delta a esa ruta (con su _delta_log/), pero NO aparece como tabla en el catálogo.

Modos de escritura (.mode())

Aplican a save() y saveAsTable():

# Sobrescribir cualquier fichero existente
df.write.format("delta").mode("overwrite").save(delta_path)

# Añadir filas nuevas a lo existente
df.write.format("delta").mode("append").save(delta_path)

Comparativa rápida de métodos

Este tema ya lo vimos en módulos anteriores, pero aquí lo vemos desde la perspectiva Delta.

Managed table

• Definición: Fabric controla metadata + datos.
• Ubicación datos: Tables/<nombre_tabla>/ en el lakehouse.
• Creación: sin especificar path.
• DROP: borra metadata Y datos.

# Managed
df.write.format("delta").saveAsTable("managed_products")

# Al dropear...
spark.sql("DROP TABLE managed_products")
# ✅ Se borra el metadata
# ✅ Se borran los ficheros Parquet + _delta_log de Tables/managed_products/

External table

• Definición: Fabric controla solo el metadata; los datos viven donde tú digas.
• Ubicación datos: la ruta que especificas con path o LOCATION.
• Creación: con path=... o LOCATION '...'.
• DROP: borra solo el metadata. Los datos permanecen.

# External
df.write.format("delta").saveAsTable(
    "external_products",
    path="Files/external_products"
)

# Al dropear...
spark.sql("DROP TABLE external_products")
# ✅ Se borra el metadata
# ❌ Los ficheros Parquet + _delta_log de Files/external_products/ permanecen

🎯 Cuando el schema se infiere

Cuando creas una external table con LOCATION apuntando a datos ya existentes:

%%sql
CREATE TABLE MyExternalTable
USING DELTA
LOCATION 'Files/mydata'

El schema se infiere de los ficheros Parquet en Files/mydata/. No hace falta especificarlo.

Esta API te permite crear tablas Delta programáticamente con control fino.

Uso básico

from delta.tables import *

DeltaTable.create(spark) \
  .tableName("products") \
  .addColumn("ProductId", "INT") \
  .addColumn("ProductName", "STRING") \
  .addColumn("Category", "STRING") \
  .addColumn("Price", "FLOAT") \
  .execute()

Con propiedades avanzadas

from delta.tables import *
from pyspark.sql.types import *

DeltaTable.create(spark) \
  .tableName("orders") \
  .addColumn("OrderId", IntegerType(), nullable=False) \
  .addColumn("OrderDate", TimestampType(), nullable=False) \
  .addColumn("CustomerId", IntegerType()) \
  .addColumn("Total", DecimalType(10,2)) \
  .addColumn("Region", StringType(), comment="Sales region") \
  .partitionedBy("Region") \
  .property("description", "Orders table with region partitioning") \
  .execute()

createIfNotExists y createOrReplace

# Solo crea si no existe (no falla si ya está)
DeltaTable.createIfNotExists(spark) \
  .tableName("products") \
  .addColumn("id", "INT") \
  .execute()

# Sobrescribe si existe
DeltaTable.createOrReplace(spark) \
  .tableName("products") \
  .addColumn("id", "INT") \
  .execute()

Cuándo usarlo vs SQL

• DeltaTableBuilder API: cuando construyes el schema programáticamente (por ejemplo, generado dinámicamente), o cuando tienes lógica condicional.
• SQL CREATE TABLE: cuando conoces el schema de antemano y es más legible en SQL.

Ambos son equivalentes en cuanto a lo que producen.

A veces quieres persistir datos en formato Delta sin registrarlos como tabla. Usos típicos:

• Datos intermedios de un ETL que no necesitas exponer.
• Datos que consumirás con la Delta API directamente por ruta.
• Datos que luego "overlayarás" con una tabla definida.

Guardar directamente

delta_path = "Files/mydatatable"

# Escritura inicial
df.write.format("delta").save(delta_path)

# Sobrescribir
new_df.write.format("delta").mode("overwrite").save(delta_path)

# Append
more_rows_df.write.format("delta").mode("append").save(delta_path)

Leer datos Delta por ruta

# Sin necesidad de que exista en el metastore
df = spark.read.format("delta").load("Files/mydatatable")
display(df)

Usar la Delta API por ruta

from delta.tables import *

# Referenciar tabla Delta por ruta
deltaTable = DeltaTable.forPath(spark, "Files/mydatatable")

# Operar sobre ella
deltaTable.update(
    condition="Category = 'Discontinued'",
    set={"Active": "false"}
)

🎯 Tip: Automatic table discovery

Si guardas datos Delta en Tables/nombre/, Fabric detecta automáticamente y crea el metadata:

# Los datos van a Files (no se registra como tabla)
df.write.format("delta").save("Files/mydatatable")   # No aparece en Tables

# Los datos van a Tables (Fabric detecta y registra automáticamente)
df.write.format("delta").save("Tables/mydatatable")  # ✨ Aparece como tabla

Aquí es donde Delta brilla: puedes hacer operaciones relacionales completas sobre tu data lake.

SELECT (leer)

Spark SQL:

%%sql
SELECT * FROM products WHERE Category = 'Bikes'

PySpark con spark.sql:

df = spark.sql("SELECT * FROM products WHERE Category = 'Bikes'")
display(df)

PySpark con read.table:

df = spark.read.table("products").filter("Category = 'Bikes'")
display(df)

INSERT (añadir)

Con SQL directo:

%%sql
INSERT INTO products VALUES (1, 'Widget', 'Accessories', 2.99)

Con embed en PySpark:

spark.sql("INSERT INTO products VALUES (1, 'Widget', 'Accessories', 2.99)")

Con append de DataFrame:

new_products_df.write.format("delta").mode("append").saveAsTable("products")

UPDATE (modificar)

Con SQL:

%%sql
UPDATE products
SET ListPrice = 2.49
WHERE ProductId = 1

Con Delta API (por tabla):

from delta.tables import *

deltaTable = DeltaTable.forName(spark, "products")
deltaTable.update(
    condition="ProductId = 1",
    set={"ListPrice": "2.49"}
)

Con Delta API (por ruta):

from delta.tables import *
from pyspark.sql.functions import *

delta_path = "Files/products"
deltaTable = DeltaTable.forPath(spark, delta_path)

# Reducir precio de accesorios un 10%
deltaTable.update(
    condition="Category == 'Accessories'",
    set={"Price": "Price * 0.9"}
)

DELETE (eliminar filas)

Con SQL:

%%sql
DELETE FROM products WHERE Discontinued = true

Con Delta API:

deltaTable = DeltaTable.forName(spark, "products")
deltaTable.delete(condition="Discontinued = true")

Ambos hacen lo mismo, pero tienen usos preferidos según el contexto.

Cuándo usar cada uno

Spark SQL (spark.sql(...) o %%sql) es mejor cuando:

• Trabajas con catalog tables (spark.sql("SELECT ...")).
• Vienes de background SQL.
• La lógica es analítica pura (SELECT, JOIN, GROUP BY).
• Quieres código más legible declarativo.

Delta API (DeltaTable.forPath() o DeltaTable.forName()) es mejor cuando:

• Trabajas con datos Delta por ruta sin tabla en el catálogo.
• Necesitas operaciones complejas programáticas (MERGE dinámico).
• Necesitas la API completa Delta (schema evolution, history, generateManifest, etc.).
• Prefieres estilo programático.

Comparativa lado a lado

UPDATE con SQL:

UPDATE products SET ListPrice = ListPrice * 0.9 WHERE Category = 'Accessories'

UPDATE con Delta API:

deltaTable.update(
    condition="Category = 'Accessories'",
    set={"ListPrice": "ListPrice * 0.9"}
)

Ambos hacen exactamente lo mismo. Elige lo que te sea más cómodo.

Cargar la referencia de la tabla

Por nombre (necesita estar en el catálogo):

deltaTable = DeltaTable.forName(spark, "products")

Por ruta (funciona incluso sin catálogo):

deltaTable = DeltaTable.forPath(spark, "Files/products")

MERGE INTO es probablemente la operación más importante de Delta Lake para casos reales. Combina INSERT, UPDATE y DELETE en una sola operación atómica basada en si las filas hacen match o no.

Sintaxis básica

MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED THEN UPDATE SET t.value = s.value
WHEN NOT MATCHED THEN INSERT (key, value) VALUES (s.key, s.value)

En cristiano:

• MERGE INTO target USING source → "Voy a fusionar filas de source en target"
• ON condition → "Considera que hacen match si..."
• WHEN MATCHED → "Cuando una fila hace match..."
• WHEN NOT MATCHED → "Cuando una fila del source NO existe en target..."
• WHEN NOT MATCHED BY SOURCE → "Cuando una fila del target NO está en source..." (útil para deletes)

Caso de uso 1: Upsert clásico

Tienes una tabla people10m (target). Recibes actualizaciones diarias en people10m_updates (source). Quieres actualizar los que existen e insertar los nuevos.

MERGE INTO people10m t
USING people10m_updates s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET *   -- Actualiza todas las columnas
WHEN NOT MATCHED THEN INSERT *   -- Inserta todas las columnas

El * funciona cuando ambos tienen las mismas columnas.

Caso de uso 2: MERGE con lógica condicional

MERGE INTO products t
USING product_updates s
ON t.ProductId = s.ProductId
WHEN MATCHED AND s.Category = 'Discontinued' THEN
    DELETE
WHEN MATCHED AND s.Price != t.Price THEN
    UPDATE SET t.Price = s.Price, t.UpdatedAt = current_timestamp()
WHEN NOT MATCHED AND s.Category IS NOT NULL THEN
    INSERT (ProductId, Name, Category, Price)
    VALUES (s.ProductId, s.Name, s.Category, s.Price)

Aquí demostramos:

• MERGE que hace DELETE si la fila está discontinued.
• MERGE que hace UPDATE solo si el precio cambió.
• MERGE que hace INSERT solo si tiene categoría.

Caso de uso 3: Deduplicación

Insertar filas nuevas de un source, ignorando las duplicadas:

MERGE INTO orders t
USING new_orders s
ON t.OrderId = s.OrderId
WHEN NOT MATCHED THEN
    INSERT *

Como no hay WHEN MATCHED, si el OrderId ya existe → no hace nada. Los duplicados se descartan silenciosamente. Muy útil para exactly-once semantics en pipelines de ingesta.

Caso de uso 4: WHEN NOT MATCHED BY SOURCE (deletes)

Quieres que target refleje exactamente lo que hay en source (útil para sincronización total):

MERGE INTO customers t
USING (SELECT * FROM customers_source WHERE updated_at >= current_date() - INTERVAL '5' DAY) s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
WHEN NOT MATCHED BY SOURCE
    AND t.updated_at >= current_date() - INTERVAL '5' DAY
    THEN DELETE

Este pattern es genial para escenarios donde el source puede cambiar los últimos N días y tú quieres reflejar exactamente esos cambios en target.

🎯 Caso de uso 5: SCD Type 1 (sobrescribir historial)

SCD Type 1: cada cambio sobrescribe el valor anterior, sin guardar historia. La tabla siempre refleja el estado actual.

MERGE INTO dim_customer t
USING staging_customer s
ON t.CustomerKey = s.CustomerKey
WHEN MATCHED THEN
    UPDATE SET
        t.Name = s.Name,
        t.Email = s.Email,
        t.State = s.State,
        t.UpdatedAt = current_timestamp()
WHEN NOT MATCHED THEN
    INSERT (CustomerKey, Name, Email, State, CreatedAt)
    VALUES (s.CustomerKey, s.Name, s.Email, s.State, current_timestamp())

Fácil, eficiente. Cuando un cliente cambia de state, se sobrescribe.

🎯 Caso de uso 6: SCD Type 2 (guardar historial)

SCD Type 2: cada cambio crea una nueva fila con validez temporal (Valid_From, Valid_To, Is_Current). La tabla guarda TODAS las versiones históricas.

Ejemplo del comportamiento:

Implementación con MERGE en dos pasos.

Paso 1: cerrar las filas actuales que han cambiado.

MERGE INTO dim_customer_scd2 t
USING (
    SELECT s.CustomerID,
           s.Name,
           s.State,
           current_date() AS effective_date
    FROM staging_customer s
) s
ON t.CustomerID = s.CustomerID AND t.Is_Current = true AND t.State != s.State
WHEN MATCHED THEN
    UPDATE SET
        t.Valid_To = s.effective_date,
        t.Is_Current = false

Paso 2: insertar las nuevas versiones.

INSERT INTO dim_customer_scd2 (CustomerID, Name, State, Valid_From, Valid_To, Is_Current)
SELECT
    s.CustomerID, s.Name, s.State,
    current_date(), NULL, true
FROM staging_customer s
LEFT JOIN dim_customer_scd2 t
    ON s.CustomerID = t.CustomerID AND t.Is_Current = true
WHERE t.CustomerID IS NULL              -- No existía
   OR t.State != s.State                -- O ha cambiado

🎯 Limitación importante

"A merge operation can fail if multiple rows of the source dataset match and the merge attempts to update the same rows of the target Delta Lake table."

Es decir: si en el source hay filas duplicadas que hacen match con la misma fila del target, MERGE falla con error (por ambigüedad semántica).

Solución: deduplicar el source antes de MERGE:

source_dedup = source_df.dropDuplicates(["id"])

SCD Type 1 vs SCD Type 2 (comparativa oficial)

Schema evolution = capacidad de modificar el schema de una tabla Delta sin romper compatibilidad.

Por qué importa

En la vida real, los schemas cambian:

• Añades columnas nuevas.
• Cambias tipos de datos.
• Renombras (menos común).

Delta permite gestionar estos cambios con seguridad.

Añadir columnas al escribir

Con mergeSchema en escritura:

# El source tiene columnas nuevas que no están en target
new_df.write.format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .saveAsTable("products")

Las columnas nuevas del source se añaden automáticamente al target. Las filas existentes tendrán NULL en esas columnas.

Sin mergeSchema → falla con error de schema mismatch (schema enforcement).

Sobrescribir schema completamente

Cuando cambias tipos o estructura radicalmente:

df.write.format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .saveAsTable("products")

Schema evolution con MERGE

Con MERGE, puedes usar WITH SCHEMA EVOLUTION:

MERGE WITH SCHEMA EVOLUTION INTO sales AS target
USING staged_sales AS source
ON target.order_id = source.order_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

Si staged_sales tiene columnas nuevas, se añaden automáticamente a sales.

Activar globalmente

Puedes activar schema evolution por sesión:

spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")

Cuando esté activo, cualquier MERGE hará schema evolution automáticamente.

ALTER TABLE (cambios explícitos)

%%sql

-- Añadir columna
ALTER TABLE products ADD COLUMNS (Discount FLOAT)

-- Cambiar comment
ALTER TABLE products ALTER COLUMN Name COMMENT 'Product name'

-- Renombrar (requiere propiedad especial habilitada)
ALTER TABLE products RENAME COLUMN OldName TO NewName

Limitación: schema enforcement por defecto

Por defecto, Delta rechaza escrituras que no cumplen el schema:

• Columna de más en source → error
• Tipo distinto → error
• Columna faltante → puede o no fallar según el modo

Uno de los superpoderes más chulos de Delta. 🌸

Ver el historial de una tabla

Por nombre:

%%sql
DESCRIBE HISTORY products

Por ruta (external tables):

%%sql
DESCRIBE HISTORY 'Files/mytable'

Resultado típico:

Ves cada versión, el timestamp, la operación y los parámetros.

Consultar una versión específica (por número)

# Leer la tabla como estaba en la versión 0
df = spark.read.format("delta") \
    .option("versionAsOf", 0) \
    .load("Files/mytable")

display(df)

O con SQL:

%%sql
SELECT * FROM products VERSION AS OF 0

Consultar por timestamp

# La tabla como estaba en enero 2026
df = spark.read.format("delta") \
    .option("timestampAsOf", "2026-01-01") \
    .load("Files/mytable")

O con SQL:

%%sql
SELECT * FROM products TIMESTAMP AS OF '2026-01-01T00:00:00'

Casos de uso reales

• 🔍 Auditoría: "¿cómo estaba esta tabla el día antes del incidente?"
• 🔄 Revertir un error: "reescribe la tabla desde la versión X"
• 📊 Análisis histórico: comparar estado actual con estado hace un mes.
• 🐛 Debugging: reproducir un bug que se dio hace días.

🎯 Rollback: revertir a una versión anterior

%%sql
-- Restaurar la tabla al estado de la versión 3
RESTORE TABLE products TO VERSION AS OF 3

O:

%%sql
RESTORE TABLE products TO TIMESTAMP AS OF '2026-04-01'

⚠️ Requiere que los ficheros de esa versión aún existan (no hayan sido purgados por VACUUM).

Puedes cambiar la retención:

ALTER TABLE products SET TBLPROPERTIES ('delta.logRetentionDuration' = 'interval 30 days')

🎯 Propiedades relacionadas

Después de estos periodos, VACUUM puede limpiar.

Tres features clave para mantener el rendimiento. Este apartado es súper examinable.

El problema: "small files problem"

Cada operación de escritura en Delta genera nuevos ficheros Parquet. Si haces muchas escrituras pequeñas (por ejemplo, streaming a alta frecuencia) → acumulas muchísimos ficheros pequeños.

Consecuencia:

• Lecturas más lentas (Spark tiene que abrir muchos ficheros).
• Queries pueden fallar por overhead.
• Costes más altos.

Delta tiene 3 features para resolverlo.

🎯 OptimizeWrite (prevención)

OptimizeWrite es una feature que al escribir, consolida datos en menos ficheros más grandes. Previene el problema antes de que ocurra.

En Fabric está activada por defecto.

# Desactivar
spark.conf.set("spark.microsoft.delta.optimizeWrite.enabled", False)

# Activar
spark.conf.set("spark.microsoft.delta.optimizeWrite.enabled", True)

# Consultar estado actual
print(spark.conf.get("spark.microsoft.delta.optimizeWrite.enabled"))

También configurable a nivel:

• Table Properties (por tabla específica)
• Individual write commands (por escritura concreta)

🎯 OPTIMIZE (mantenimiento)

OPTIMIZE es un comando de mantenimiento que consolida ficheros Parquet pequeños en ficheros más grandes.

Cuándo ejecutar:

• Después de cargar tablas grandes.
• Periódicamente en tablas con muchas escrituras.
• Cuando notas queries lentas.

Beneficios: menos ficheros, mejor compresión, mejor distribución en nodos y queries más rápidas.

Cómo ejecutar (3 formas):

A) Desde el Lakehouse Explorer (UI): menú ... junto a la tabla → Maintenance → Run OPTIMIZE command → opcionalmente marcar V-Order → Run now.

B) Desde SQL:

%%sql
OPTIMIZE products

C) Con Z-ORDER (co-localizar datos por columna):

%%sql
OPTIMIZE products ZORDER BY (Category)

ZORDER es una técnica avanzada que reordena los datos dentro de los ficheros para acelerar filtros por esa columna.

🎯 V-Order (aceleración de lecturas)

V-Order es una feature exclusiva de Fabric que optimiza los ficheros Parquet para lecturas ultra rápidas desde motores de Fabric.

Características clave:

• ✅ Activada por defecto en Fabric.
• ⚡ Habilita "lightning-fast reads" con acceso in-memory-like.
• 💰 Reduce recursos de red, disco y CPU en lecturas.
• 📈 Aplicada al escribir datos.
• 🐢 Overhead pequeño (~15%) en escritura, gran beneficio en lectura.
• 📊 Compatible con motores externos (100% Parquet compliant).

Cómo funciona técnicamente: aplica sorting especial, distribución de row groups optimizada, dictionary encoding y compresión.

Efecto en distintos motores:

Cuándo desactivarlo: en write-intensive scenarios como staging bronze donde los datos se leen 1-2 veces solo, el overhead del 15% no se compensa. En esos casos, desactivar V-Order acelera la ingesta.

Aplicar V-Order con OPTIMIZE: desde la UI en Table Maintenance → marcar checkbox "Apply V-order"; o automáticamente al escribir (si está activo por defecto).

Comparativa de las 3 features

¿Qué hace VACUUM?

VACUUM elimina físicamente los ficheros Parquet obsoletos de una tabla Delta:

• Ficheros que ya no están referenciados en el transaction log.
• Ficheros más antiguos que el retention period.

IMPORTANTE: VACUUM NO borra el transaction log, solo los ficheros de datos. El historial de operaciones sigue en el _delta_log/.

Retention period

Default: 7 días (168 horas).

El sistema impide por defecto usar un retention menor a 7 días, porque puede causar problemas:

• Rompe time travel.
• Puede corromper lecturas concurrentes (readers largos que aún referencian ficheros que se están borrando).

Cómo ejecutar VACUUM

Desde el Lakehouse Explorer: menú ... junto a la tabla → Maintenance → Run VACUUM command using retention threshold → configurar threshold → Run now.

Con SQL:

%%sql
VACUUM products RETAIN 168 HOURS

Con lakehouse.tabla:

%%sql
VACUUM lakehouse2.products RETAIN 168 HOURS

Con retention custom (más de 7 días, más seguro):

%%sql
VACUUM products RETAIN 720 HOURS   -- 30 días

⚠️ Cómo bajar el retention (peligroso)

Si REALMENTE necesitas bajar de 7 días (raramente):

# Desactivar protección
spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")

# Ahora puedes:
spark.sql("VACUUM products RETAIN 24 HOURS")

Ver historial de VACUUM

VACUUM se registra en el transaction log:

%%sql
DESCRIBE HISTORY products

Verás entradas con operation = VACUUM START y VACUUM END.

Cuándo ejecutar VACUUM

• Semanalmente/mensualmente en tablas con muchas modificaciones.
• Después de grandes DELETE/UPDATE en tablas grandes.
• Cuando el storage crece sin explicación (ficheros obsoletos acumulados).
• Como parte de un maintenance schedule automatizado.

Cómo elegir el retention

Basado en: requisitos de retención de datos (compliance), tamaño y coste de storage, frecuencia de cambios y requisitos regulatorios (GDPR, HIPAA).

Ya vimos particionado en el Módulo 3, aquí lo profundizamos con específicos Delta.

Recordatorio: qué es

Particionar = dividir el output en subcarpetas basadas en el valor de una columna:

Files/sales/
├── year=2024/
│   └── part-XXX.parquet
├── year=2025/
│   └── part-XXX.parquet
└── year=2026/
    └── part-XXX.parquet

Beneficio principal: data skipping

Cuando filtras por la columna particionada, Delta hace partition pruning: solo lee las particiones relevantes.

# Delta lee solo la partición year=2026
df = spark.read.format("delta").table("sales").filter("year = 2026")

Particionar al crear (Spark)

df.write.format("delta") \
    .partitionBy("Category") \
    .saveAsTable("partitioned_products", path="Files/partitioned_products")

Genera:

Files/partitioned_products/
├── Category=Bikes/
├── Category=Accessories/
├── Category=Clothing/
└── _delta_log/

Particionar al crear (SQL)

%%sql
CREATE TABLE partitioned_products (
    ProductID INTEGER,
    ProductName STRING,
    Category STRING,
    ListPrice DOUBLE
)
PARTITIONED BY (Category)

🎯 Cuándo particionar

SÍ particionar cuando:

• Tienes muchos datos (cientos de GB o más).
• Puedes dividir en pocas particiones grandes (no miles).
• Las queries filtran frecuentemente por esa columna.

NO particionar cuando:

• Los volúmenes son pequeños (< 1GB total).
• La columna tiene alta cardinalidad (muchos valores únicos).
• Necesitarías múltiples niveles de particionado que no aportan.

Limitación importante

Las particiones son un layout fijo. Cuando decides particionar por year, esa decisión afecta todos los patrones de query. No se adapta dinámicamente.

Alternativas modernas para agilidad:

• Liquid Clustering (feature reciente de Delta) — más flexible.
• Z-ORDER con OPTIMIZE — reorganiza sin cambiar layout físico.

Trampa (repaso)

Cuando lees directamente una partición:

road_bikes = spark.read.parquet("Files/bike_data/Category=Road Bikes")

La columna Category desaparece del DataFrame (Spark ya sabe que todos son "Road Bikes"). Si haces road_bikes.columns, no verás Category.

Una de las características más potentes de Delta: se integra con Spark Structured Streaming para casos de streaming de datos.

¿Qué es Spark Structured Streaming?

Es una API de Spark para procesar streams de datos en tiempo real. La abstracción clave: un stream es un DataFrame ilimitado que va creciendo.

Con Spark Structured Streaming puedes leer de:

• Ports de red.
• Sistemas de mensajería: Azure Event Hubs, Kafka.
• Ubicaciones de file system (poll files).
• Delta tables ← la que nos interesa.

Y escribir a Delta tables ← la otra que nos interesa, y muchas otras.

Delta como sink (destino de un stream)

Caso: capturar telemetría IoT y guardarla en una tabla Delta.

# Leer stream desde una fuente (por ejemplo Event Hubs)
stream_df = spark.readStream.format("eventhubs") \
    .options(**event_hub_conf) \
    .load()

# Escribir a Delta
query = stream_df.writeStream \
    .format("delta") \
    .option("checkpointLocation", "Files/checkpoints/iot_stream") \
    .start("Tables/iot_readings")

# Verificar
print("Streaming to iot_readings...")

Delta como source (leer una tabla como stream)

Caso: procesar en real time los cambios de una tabla Delta.

# Leer una tabla Delta como stream
stream_df = spark.readStream.format("delta") \
    .option("ignoreChanges", "true") \
    .table("orders_in")

# Verificar que es streaming
print(stream_df.isStreaming)   # True

Ejemplo end-to-end completo

Paso 1: crear tabla source

%%sql
CREATE TABLE orders_in (
    OrderID INT,
    OrderDate DATE,
    Customer STRING,
    Product STRING,
    Quantity INT,
    Price DECIMAL
) USING DELTA

Paso 2: insertar datos hipotéticos

%%sql
INSERT INTO orders_in VALUES
    (3001, '2026-09-01', 'Yang', 'Road Bike Red', 1, 1200),
    (3002, '2026-09-01', 'Carlson', 'Mountain Bike Silver', 1, 1500),
    (3003, '2026-09-02', 'Wilson', 'Road Bike Yellow', 2, 1350),
    (3004, '2026-09-02', 'Yang', 'Road Front Wheel', 1, 115),
    (3005, '2026-09-02', 'Rai', 'Mountain Bike Black', 1, NULL)

Paso 3: leer como stream

stream_df = spark.readStream.format("delta") \
    .option("ignoreChanges", "true") \
    .table("orders_in")

Paso 4: transformar el stream

from pyspark.sql.functions import col, expr

# Filtrar precios nulos y añadir columnas calculadas
transformed_df = stream_df \
    .filter(col("Price").isNotNull()) \
    .withColumn('IsBike', expr("INSTR(Product, 'Bike') > 0").cast('int')) \
    .withColumn('Total', expr("Quantity * Price").cast('decimal'))

Paso 5: escribir el stream a otra tabla Delta

output_table_path = 'Tables/orders_processed'
checkpoint_path = 'Files/delta/checkpoint'

deltastream = transformed_df.writeStream \
    .format("delta") \
    .option("checkpointLocation", checkpoint_path) \
    .start(output_table_path)

print("Streaming to orders_processed...")

Paso 6: consultar el resultado

%%sql
SELECT * FROM orders_processed ORDER BY OrderID

Resultado esperado (nota: order 3005 excluido por NULL en Price):

Paso 7: parar el stream (siempre importante)

deltastream.stop()

🎯 El famoso checkpointLocation

Este parámetro es absolutamente crítico en streaming:

.option("checkpointLocation", "Files/checkpoints/mi_stream")

Qué hace:

• Guarda el estado del procesamiento del stream.
• Registra qué offset del source se ha procesado ya.
• Permite recuperarse de fallos: si el stream se cae, puede continuar desde donde se quedó.

Reglas:

• Cada stream debe tener su propio checkpoint location único.
• Nunca reutilices un checkpoint entre streams distintos.
• Nunca lo borres mientras el stream esté activo.

Modos de salida (output modes)

Al escribir un stream, especificas outputMode:

.writeStream.outputMode("append").format("delta")...

MERGE en streaming: foreachBatch

Para hacer MERGE (upsert) desde un stream, usas foreachBatch:

def merge_batch(microbatch_df, batch_id):
    # microbatch_df es un DataFrame estándar (no streaming)
    # Puedes hacer MERGE aquí
    microbatch_df.createOrReplaceTempView("updates")

    spark.sql("""
        MERGE INTO target t
        USING updates s
        ON t.id = s.id
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)

# Aplicar la función a cada microbatch del stream
stream_df.writeStream \
    .foreachBatch(merge_batch) \
    .option("checkpointLocation", "Files/checkpoints/merge_stream") \
    .start()

Casos de uso reales

• 📱 IoT / telemetría: dispositivos → Event Hubs → Structured Streaming → Delta.
• 📊 CDC (Change Data Capture): cambios de BBDD → Kafka → Streaming → Delta con MERGE.
• 🌐 Clickstream analytics: eventos web → tabla Delta → dashboards en tiempo casi real.
• 📝 Log processing: aplicaciones → logs → Delta particionada por hora.

• Trampa 1: "Al hacer UPDATE, se modifica el fichero Parquet en sitio" ❌ Los Parquet son inmutables. UPDATE genera nuevos ficheros y marca los viejos como removidos en el _delta_log.
• Trampa 2: "Time travel funciona para siempre" ❌ Solo mientras los ficheros viejos existan. Después de VACUUM más allá del retention, ya no.
• Trampa 3: "Puedo hacer VACUUM RETAIN 0 HOURS para limpiar todo" ⚠️ El sistema te lo impide por defecto (retention < 7 días bloqueado). Y hacerlo puede corromper lecturas concurrentes y romper time travel.
• Trampa 4: "OPTIMIZE incluye VACUUM" ❌ Son operaciones distintas. OPTIMIZE consolida ficheros pequeños en grandes. VACUUM borra ficheros obsoletos.
• Trampa 5: "V-Order hay que activarlo manualmente" ❌ Está activado por defecto en Fabric. Solo lo desactivas explícitamente si tienes motivos (staging heavy-write).
• Trampa 6: "OptimizeWrite = OPTIMIZE" ❌ Son distintos: OptimizeWrite es preventivo, al escribir (activo por defecto); OPTIMIZE es un comando de mantenimiento post-hoc (manual/scheduled).
• Trampa 7: "MERGE puede aceptar filas duplicadas en el source" ❌ Si hay filas duplicadas en el source que hacen match con la misma fila del target → MERGE falla. Deduplica el source antes.
• Trampa 8: "En streaming, no necesito checkpointLocation" ❌ Es crítico para recuperación tras fallo y para exactly-once semantics.
• Trampa 9: "Puedo compartir el mismo checkpoint entre streams" ❌ NO. Cada stream necesita su propio checkpoint location único.
• Trampa 10: "Un stream desde Delta puede procesar UPDATEs del source" ❌ Por defecto solo procesa appends. Para UPDATE/DELETE del source, activa ignoreChanges o ignoreDeletes.
• Trampa 11: "Schema evolution es automático siempre" ❌ Por defecto Delta enforcea schema (rechaza escrituras que no cumplen). Schema evolution hay que activarlo explícitamente con mergeSchema=true o config.
• Trampa 12: "Particionar por columna de alta cardinalidad mejora performance" ❌ Genera miles de ficheros pequeños → peor performance. Usa baja cardinalidad.
• Trampa 13: "DROP TABLE de una external borra los datos" ❌ Solo borra el metadata. Los datos permanecen en la ubicación especificada.
• Trampa 14: "MERGE WITH SCHEMA EVOLUTION añade cualquier columna del source" ✅ Sí, si tiene columnas nuevas se añaden. Los tipos deben ser compatibles.
• Trampa 15: "V-Order es propietario de Microsoft y no lo puede leer nadie más" ❌ V-Order es 100% Parquet compliant. Cualquier motor Parquet puede leerlo (Databricks, Snowflake, etc.). Solo que los motores de Fabric (VertiScan) lo aprovechan al máximo.

OptimizeWrite vs OPTIMIZE vs V-Order

Managed vs External tables

SQL SELECT vs Delta API

SCD Type 1 vs Type 2

Delta como source vs sink de stream

Retention y VACUUM

• Módulo Learn: work-delta-lake-tables-fabric
• Lakehouse and Delta tables: lakehouse-and-delta-tables
• Delta Lake schema evolution: delta-lake-schema-evolution
• MERGE INTO syntax: delta-merge-into
• Upsert into Delta with MERGE: delta/merge
• Delta streaming reads and writes: structured-streaming/delta-lake
• CDC in Copy Job (SCD Type 1 vs 2): cdc-copy-job
• Slowly changing dimension type 2 tutorial: slowly-changing-dimension-type-two
• Drop Delta tables and schemas: delta-lake-drop
• Delta Lake documentation (open source): docs.delta.io