Ingestión en AWS: Auto Loader, file events, Kafka/MSK y sinks idempotentes

Todo lo que sigue en este curso, las transacciones Delta de la lección 04, los pipelines declarativos de la lección 07, los controles de calidad de la lección 10, asume una tabla bronze que ya se está rellenando correctamente. La ingesta es donde los proyectos de lakehouse fallan en silencio: un checkpoint borrado «para arreglar el retraso», un directorio de esquema aparcado dentro del prefijo de aterrizaje para que el stream descubra sus propios archivos de estado, una consulta de Kafka dejada en startingOffsets por defecto de modo que la primera ejecución se saltó tres horas de backlog, o un stream de listado de directorios con un trigger de un segundo cuya factura de LIST crece con cada hora de actividad.

La mecánica no es complicada. La recomendación de la propia Databricks es una frase: «Databricks also recommends Auto Loader whenever you use Apache Spark Structured Streaming to ingest data from cloud object storage» (docs.../auto-loader, última actualización 2026-09-11). Lo complicado es la higiene: dónde vive el estado, qué pasa cuando aparece una columna, qué modo de descubrimiento se paga solo, y cómo un micro-batch reintentado evita escribir las filas dos veces. Eso es esta lección.

Todos los datos que siguen proceden de la documentación de AWS de Databricks (páginas de Auto Loader, directory-listing, file-notification, esquema, producción, clean-source, mejores prácticas, FAQ, checkpoints, output-mode, triggers, foreach, Delta streaming, Kafka y JDBC), de la documentación de Apache Spark y de AWS, y de blogs de ingeniería de Databricks fechados: todos consultados o fechados en octubre de 2026. El alcance de la plataforma es Databricks sobre AWS: Azure Event Hubs no existe sobre este sustrato y queda fuera de aquí.

En esta lección vas a:

• escribir el propio notebook bronze de Auto Loader, cloudFiles.format, schemaLocation, checkpointLocation, useManagedFileEvents, inferColumnTypes, schemaEvolutionMode, mergeSchema en el sink Delta, y narrar cada valor por defecto;
• situar la ubicación de esquema y el checkpoint donde no puedan ser descubiertos, caducados ni compartidos, y defender esa elección frente al instinto de «poner el estado junto a los datos»;
• calcular las tres facturas de LIST para un prefijo de aterrizaje bajo tres políticas de trigger, y elegir entre listado de directorios, notificaciones de archivos clásicas SNS+SQS y file events gestionados con sus permisos, límites y comportamiento de backfill;
• ejecutar un boceto de rendimiento para un pico de 1,8 millones de archivos: tasa de llegada frente a tasa de servicio, el límite duro de archivos frente al límite blando de bytes, y qué pasa cuando alguien «con cuidado» deja maxFilesPerTrigger demasiado bajo;
• configurar Amazon MSK con AWS_MSK_IAM, incluidos los nombres de clase con shaded y el puerto que debe coincidir con el modo de autenticación, más la aritmética de dimensionado de maxOffsetsPerTrigger;
• elegir un output mode contra la matriz de soporte de Delta, hacer idempotente foreachBatch con txnAppId/txnVersion, y defender la cadena exactly-once de extremo a extremo;
• unificar batch y streaming en una sola ruta de código con Trigger.AvailableNow, y calcular el coste del job bronze nocturno en DBUs a octubre de 2026;
• estudiar dos casos reales de clientes de AWS (Asurion y Burberry), leer el recuadro de actualizaciones de 2026 y practicar con 13 preguntas de estilo examen más cuatro comprobaciones interactivas.

---

1. Ingestión en el lakehouse de AWS: tres familias de origen, una ruta de escritura

La ingesta sobre Databricks en AWS tiene tres familias que importan en producción, más un juguete:

Tres cosas de esa tabla son estructurales. Primero, las cuatro convergen en la misma ruta de escritura: writeStream → checkpointLocation → output mode → Delta. El código de transformación es idéntico para spark.read y spark.readStream porque ambos devuelven DataFrames, esa es la historia de la unificación batch/streaming de la sección 9. Segundo, JDBC no tiene ninguna fuente de streaming documentada por Databricks: la documentación actual describe lecturas por lotes, conexiones UC gobernadas y foreachBatch para las escrituras; el DML a nivel de fila sobre la base de datos de origen no está soportado explícitamente («The Spark JDBC data source does not support arbitrary DML statements», docs.../connect/jdbc-connection, consultado 2026-10). Tercero, el checkpoint es la consulta, bórralo y arrancas de cero: para Kafka eso significa que startingOffsets vuelve a aplicarse; para Auto Loader, que el estado de archivos descubiertos desapareció.

• 📚 ¿Sabías que…? Auto Loader no es nuevo. Databricks lo anunció el 2020-02-24 como «an optimized cloud file source for Apache Spark that loads data continuously and efficiently from cloud storage as new data arrives», nombrando a Fivetran, Qlik, Infoworks, StreamSets y Syncsort como herramientas de ingesta socias en el lanzamiento (databricks.com/blog/2020/02/24/introducing-databricks-ingest-easy-data-ingestion-into-delta-lake.html, 2020-02-24). El problema que resolvió nunca cambió: «Tanto el coste como la latencia pueden sumarse rápidamente a medida que se añaden más y más archivos a un directorio debido al listado repetido de archivos.» Las cifras de escala que verás citadas, «billions of files» para backfills y «millions of files per hour» para ingesta casi en tiempo real, son afirmaciones de capacidad del proveedor de la página de docs (2026-09-11), no benchmarks de clientes: trátalas como margen de diseño, nunca como un SLA.

---

2. Anatomía de Auto Loader: la fuente cloudFiles

2.1 Las opciones que importan (oct 2026)

spark.readStream.format("cloudFiles") es una fuente ordinaria de Structured Streaming. Lo que lo hace Auto Loader es el motor de descubrimiento: a medida que se encuentran archivos, sus metadatos se persisten «in a scalable key-value store (RocksDB) in the checkpoint location», lo que «ensures that data is processed exactly once» (docs .../auto-loader, 2026-09-11).

Dos valores por defecto sorprenden a todo el mundo. includeExistingFiles=true significa que el primer arranque de un checkpoint nuevo recorre todo el prefijo (eso es tu backfill). Y maxFilesPerTrigger=1000 es un límite duro mientras que el límite de bytes es blando: «An individual file is never split across micro-batches» (docs.../spark/api-options, fecha de página 2026-10-05).

2.2 Ejemplo resuelto E1: un notebook bronze de Auto Loader con forma de producción

Lo que significan los números mientras se ejecuta: el primer arranque muestrea los primeros 50 GB o 1.000 archivos (lo que se cruce primero, los más recientes por hora de modificación) para inferir el esquema; cada micro-batch posterior está limitado a 1.000 archivos salvo que el runtime lo auto-ajuste (DBR 18.0+); Trigger.AvailableNow procesa solo los archivos que llegaron antes de la hora de inicio de la consulta y se apaga solo; y si aparece una columna nueva, la consulta se detiene con UnknownFieldException después de que _schemas ya se haya actualizado: así que el job, y no una persona, la reinicia (sección 3).

---

3. Inferencia, evolución y rescue de esquemas

3.1 Qué muestrea realmente la inferencia

En la primera definición del DataFrame, Auto Loader «lists your source directory and chooses the most recent (by file modification time) 50 GB of data or 1000 files, and uses those to infer your data schema» (FAQ de Auto Loader, fechado 2026-06-23). El resultado se escribe en un directorio _schemas bajo cloudFiles.schemaLocation, y es el historial de esquema que el stream recarga en cada reinicio: «Cuando el stream se reinicia, toma el esquema evolucionado de la ubicación de esquema y comienza a ejecutarse sin ninguna sobrecarga de la inferencia» (misma FAQ).

Valores por defecto que muerden:

• JSON, CSV e XML infieren todas las columnas como strings salvo que fijes cloudFiles.inferColumnTypes=true. Los formatos con tipos (Parquet, Avro) fusionan en su lugar los tipos codificados.
• Si el directorio de origen está vacío, debes suministrar un esquema: no hay nada que muestrear.
• cloudFiles.schemaHints te permite fijar tipos conocidos ("order_id=STRING, amount=DECIMAL(12,2)") sin escribir a mano el esquema completo.
• Las columnas de partición se infieren de rutas al estilo Hive, pero las columnas de partición nuevas no se evolucionan: decláralas con cloudFiles.partitionColumns.
• Existen overrides: spark.databricks.cloudFiles.schemaInference.sampleSize.numBytes (p. ej. 10gb) y ...sampleSize.numFiles.

3.2 Modos de evolución y el bucle de reinicio

La secuencia ante un cambio de esquema es fija, y vale la pena memorizarla, porque es un diseño de alerta y reproducción y no una corrección silenciosa:

Dos hechos de apoyo. El lado de escritura debe estar de acuerdo: .option("mergeSchema", "true") en el sink Delta, o la tabla rechaza el cambio aditivo. Y un renombrado se trata como una adición: el nombre nuevo se convierte en columna mientras la antigua se rellena con NULL en las filas nuevas (docs.../data-engineering/schema-evolution, consultado 2026-10). El esquema de una consulta de streaming está «locked in during the planning phase», así que cualquier cambio de fuente a mitad de vuelo hace fallar la consulta y obliga a reiniciar; por eso las docs recomiendan ejecutar los streams de Auto Loader como jobs de Lakeflow/Workflows y no como notebooks all-purpose que corren para siempre.

3.3 Ejemplo resuelto E2: un stream de eventos que nunca falla con tipos fijados

_rescued_data guarda lo que el esquema no pudo aceptar, «column is missing from the schema / type mismatches / case mismatches», como un blob JSON junto a la ruta del archivo de origen, y así mantienes a la vez un flujo sin pérdidas y una señal de calidad medible. Usa rescue en los flujos que nunca deben detenerse (clickstream, IoT); usa addNewColumns más un job con reinicio automático en los flujos donde una columna nueva es un evento que quieres notar.

3.4 ⚠️ ADVERTENCIA: nunca pongas el estado de esquema ni el checkpoint en la ruta de datos

⚠️ ADVERTENCIA: cloudFiles.schemaLocation y checkpointLocation nunca deben vivir dentro del directorio de aterrizaje de origen, y tampoco bajo una política de ciclo de vida de objetos en la nube.

1. Estar en la ruta de datos significa autodescubrimiento. Si _schemas/ o el checkpoint viven bajo s3://acme-landing/raw/orders/, Auto Loader terminará listando sus propios archivos de estado como entradas. Mantén el estado en un prefijo distinto o en un volumen de Unity Catalog (/Volumes/<catalog>/<schema>/checkpoints/...), nunca bajo el directorio de datos de la tabla.
2. Una regla de ciclo de vida que caduca el estado destruye la consulta. «Databricks recommends setting the checkpoint location to a location without a cloud object lifecycle policy. If files in the checkpoint location are cleaned according to the policy, the stream state is corrupted. If this happens, you must restart the stream from scratch» (docs .../auto-loader/production, pie de página 2026-09-18). La misma frase aplica a schemaLocation.
3. Sepáralos por carga de trabajo. Cada carga de trabajo de ingesta con Auto Loader requiere un checkpoint de streaming propio. Dos cargas de trabajo compartiendo un checkpoint, o un directorio _schemas, lucharán por el estado.
4. El estado junto a una tabla Delta debe llevar prefijo de guion bajo. Si debes colocar un checkpoint junto a una tabla, usa /_checkpoints: VACUUM «omite cualquier directorio que empiece por _» (docs .../structured-streaming/delta-lake, fechado 2026-08-24), así el mantenimiento no se lo comerá.
5. La igualdad está bien, la contención no. Las docs permiten explícitamente schemaLocation == checkpointLocation; lo que nunca permiten es estado dentro de la ruta de entrada ni bajo una caducidad.

---

4. Detección de archivos: listado de directorios frente a notificaciones de archivos

Auto Loader «supports two file detection modes. By default, Auto Loader uses directory listing mode. However, Databricks recommends file notification mode using file events for most workloads» (docs .../auto-loader, 2026-09-11). Elegir entre uno y otro es la decisión de coste e infraestructura de esta lección.

4.1 Listado de directorios (por defecto)

Cada trigger lista el árbol de entrada. Dos propiedades lo hacen viable: Auto Loader «uses native cloud APIs to get lists of files», y en S3 cada LIST devuelve hasta 1.000 claves, de modo que el número de llamadas a la API escala como archivos ÷ 1.000 y no como directorios × triggers. Esa es la diferencia frente al file source de OSS Spark, cuyo coste la documentación modela como «1 (base directory) + 365 (per day) × 24 (per hour) = 8761 calls» para un layout YYYY/MM/DD/HH (docs .../auto-loader/directory-listing-mode, 2026-09-11).

La advertencia documentada trata de la frecuencia: con triggers continuos como Trigger.ProcessingTime, el modo de listado de directorios «continuously lists the entire directory and can significantly increase LIST API costs». Tres reglas de higiene:

• conserva el nomenclatura léxica de archivos (YYYY/MM/DD/HH, con ceros a la izquierda: los archivos CDC de AWS DMS como LOAD00000001.csv y los nombres de objetos de Kinesis Firehose son los ejemplos de las propias docs), aunque Auto Loader no garantiza el orden en que los archivos se descubren o se procesan;
• prefiere Trigger.AvailableNow a un trigger continuo en este modo;
• el listado incremental está obsoleto: en modo auto se ejecuta un listado completo de directorios tras 7 listados incrementales consecutivos (ajustable con cloudFiles.backfillInterval), y DBR 11.3 LTS+ te deja cambiar la ruta de entrada sin un checkpoint nuevo: una capacidad no soportada para las notificaciones de archivos.

4.2 Notificaciones de archivos clásicas: SNS + SQS

En este modo Auto Loader conecta la notificación del bucket de S3 (ObjectCreated) a través de un tema de AWS SNS hasta una cola de AWS SQS (prefijo de recurso databricks-auto-ingest) y después consume la cola. La política de configuración necesita el conjunto de acciones documentado, s3:GetBucketNotification, s3:PutBucketNotification, sns:CreateTopic, sns:Subscribe, sns:Publish, sqs:CreateQueue, sqs:ReceiveMessage, sqs:DeleteMessage, sqs:SendMessage, sqs:SetQueueAttributes, sqs:ChangeMessageVisibility, sqs:PurgeQueue sobre databricks-auto-ingest-*, y una vez configurado, el stream en ejecución solo necesita la política reducida de consumo (docs.../auto-loader/file-notification-mode, pie de página 2026-10-07). La misma página limita el patrón a 100 por bucket de S3.

Dos reservas viajan con él. La entrega de eventos es best-effort, «Cloud providers do not guarantee 100% delivery of all file events», así que los SLA de completitud en el modo clásico se compran con re-listados asíncronos de cloudFiles.backfillInterval que ingieren solo los archivos perdidos (sin duplicados, porque el checkpoint ya sabe lo que vio). Y en este modo no puedes cambiar la ruta de entrada sin un checkpoint nuevo.

4.3 File events gestionados (recomendado)

La forma moderna se activa en la external location, no por stream: una sola cola de notificación de archivos para todos los streams que procesan archivos de una external location dada, activada con cloudFiles.useManagedFileEvents=true. Requisitos: un workspace habilitado para Unity Catalog, permiso para crear un storage credential y una external location, y Databricks Runtime 14.3 LTS o superior (misma página, 2026-10-07).

Lo que obtienes: menos concesiones IAM (permiso en la nube de una vez, y después ninguna cola por stream), auto-ajuste y limpieza automáticos, backfill automático, «Databricks rellena automáticamente esas external locations con un listado completo cuando los file events se activan por primera vez, y después continúa haciéndolo aproximadamente cada 24 horas», y la instrucción explícita de no fijar un intervalo de backfill en este modo. Lo que debes respetar: «The file events service optimizes file discovery by caching the most recently created files. If Auto Loader runs infrequently, this cache can expire … invoke Auto Loader at least once every seven days.» Un job AvailableNow nocturno u horario pasa ese listero con holgura; una ejecución manual trimestral, no.

Cambiar de modo es seguro: «puedes cambiar entre notificaciones de archivos y listado de directorios en cualquier momento y seguir manteniendo las garantías de procesamiento de datos exactly-once» (misma página).

4.4 Ejemplo resuelto E3: la factura de LIST, calculada de tres formas

Entradas (todas oficiales): S3 devuelve 1.000 claves por LIST; «Las solicitudes LIST de cualquier clase de almacenamiento se facturan a la misma tarifa que las solicitudes PUT, COPY y POST de S3 Standard»: $0.005 por cada 1.000 en us-east-1, mientras que DELETE y CANCEL son gratuitos (aws.amazon.com/s3/pricing, consultado 2026-10). Prefijo de aterrizaje: 2.000.000 de archivos. La aritmética de abajo es nuestra; las entradas son de AWS.

Lee la forma, no solo los dólares: el mismo prefijo de 2M de archivos cuesta 900× más con un trigger de un segundo que con un AvailableNow de 15 minutos, porque el multiplicador es la frecuencia de listado: no el volumen de datos. Los file events gestionados sustituyen el listado por trigger por llamadas a la API de colas, dejando solo el listado automático de backfill de ~24 horas. Aquí deliberadamente no precio el lado SQS/SNS: las fuentes de esta lección no traen ninguna cifra verificada de precio de colas de AWS, así que comprueba tú mismo aws.amazon.com/pricing antes de modelarlo.

• 📚 ¿Sabías que…? LIST se factura como una escritura, no como una lectura: «Las solicitudes LIST de cualquier clase de almacenamiento se facturan a la misma tarifa que las solicitudes PUT, COPY y POST de S3 Standard»: $0.005 por 1.000: mientras que DELETE y CANCEL son gratuitos (aws.amazon.com/s3/pricing, consultado 2026-10). Y en S3 no existe un rename de propósito general: «S3 Express One Zone es la única clase de almacenamiento que soporta la API RenameObject», así que un archivo estilo MOVE es un par copy + delete facturado a las tarifas de PUT/COPY/LIST, y, en términos de notificaciones clásicas, esa copia aparece como aún otro evento ObjectCreated. Esa es una segunda razón por la que los file events gestionados, que no se disparan de nuevo con tus propias copias de contabilidad, son el modo recomendado.

---

5. Rendimiento e higiene en producción

5.1 Ejemplo resuelto E4: un boceto de rendimiento para un pico de 1,8 millones de archivos

Escenario: un socio deja 1.800.000 archivos JSON de 120 KB en el prefijo de aterrizaje durante una ventana nocturna de dos horas (216 GB), y después sigue llegando un hilo constante. Tasa de llegada = 1.800.000 ÷ 7.200 s = 250 archivos/s.

Dimensionado de los micro-batches:

• Por defecto maxFilesPerTrigger=1000 → cada micro-batch lleva 1.000 archivos = 120 MB.
• Si en su lugar fijas maxBytesPerTrigger="500mb", ese tope de bytes blando permitiría 500 MB ÷ 120 KB ≈ 4.166 archivos, pero el límite de archivos es duro, así que el batch efectivo sigue en 1.000 archivos. Gana el límite que se alcance primero.
• Supón que el cluster sostiene 1,0 s por micro-batch → tasa de servicio 1.000 archivos/s (120 MB/s). Utilización ρ = 250 ÷ 1.000 = 0,625: la cola se vacía.
• El backlog de 1,8 M de archivos se drena a 1.000 − 250 = 750 archivos/s netos → 2.400 s = 40 minutos, habiendo procesado 2,4 M de archivos en total (≈288 GB).

Ahora el modo de fallo que crea un ingeniero «cuidadoso»: dejar maxFilesPerTrigger=100 «para no forzar el cluster», lo que limita el servicio a 100 archivos/s < 250 archivos/s de llegada, la cola crece a 150 archivos/s sin límite, y ningún tamaño de cluster arregla un mando de control de admisión mal puesto. En DBR 18.0+ tanto cloudFiles.maxFilesPerTrigger como cloudFiles.maxBytesPerTrigger se «configuran de forma dinámica y no es necesario fijarlos manualmente» (docs.../spark/api-options, 2026-10-05); fijarlos a mano ahí pelea contra el auto-ajuste. Y si el pico es genuinamente enorme y de una sola vez, ejecuta Trigger.AvailableNow como tarea de Workflows en lugar de un trigger continuo, limita la tasa entre micro-batches y se apaga cuando se pone al día.

5.2 Retención: maxFileAge, cleanSource, backfill

La economía de cleanSource es el mismo argumento de la sección 4: «Archiva los archivos procesados con cloudFiles.cleanSource … Esto reduce tanto los costes de almacenamiento como los costes de listado de directorios para streams de larga duración»: dejas de pagar por re-listar archivos que nunca más leerás.

5.3 Monitorización de la fuente

Tres métricas te dicen si la ingesta está sana antes de que disparen las alertas de frescor aguas abajo: numFilesOutstanding, numBytesOutstanding y (notificaciones de archivos) approximateQueueSize: más la tabla de estado:

Un numFilesOutstanding creciente con una tasa de procesamiento estable significa que estás dimensionado por debajo (arregla los límites de tasa, no el checkpoint); un approximateQueueSize estancado en el modo clásico de notificaciones significa que el consumidor no puede vaciar la cola: permisos, o el límite de 100 por bucket.

---

6. Kafka y Amazon MSK como fuente

6.1 Las opciones que importan

Lectura mínima: spark.readStream.format("kafka").option("kafka.bootstrap.servers", "<server:9094>").option("subscribe", "orders").option("startingOffsets", "latest").load(), y después las cuatro opciones que deciden si tu primer día sale bien:

El esquema de fila es fijo, siete columnas: key binary, value binary, topic string, partition int, offset long, timestamp timestamp, timestampType int. La autenticación en AWS es SSL o SASL_SSL con PLAIN/SCRAM/AWS_MSK_IAM, o, en la ruta gobernada, un service credential / conexión de Unity Catalog, en cuyo caso no debes fijar tú kafka.security.protocol ni sasl.*. Si conduces el cliente directamente: prefija siempre las propiedades del cliente con kafka. y usa siempre los nombres de clase con shaded de Databricks (kafkashaded., shadedmskiam.), o te encontrarás con SaslException / RESTRICTED_STREAMING_OPTION_PERMISSION_ENFORCED.

6.2 Ejemplo resuelto E5: lectura MSK IAM con limitación de tasa

Aritmética de dimensionado (nuestra, sobre semánticas oficiales): la llegada pico es de 12.000 msg/s en 6 particiones a 400 bytes de media. Con processingTime="30 seconds" y maxOffsetsPerTrigger=300000, la capacidad es 300.000 ÷ 30 = 10.000 msg/s < 12.000 msg/s, el backlog crece a 2.000 msg/s. Súbelo a 600.000 → 20.000 msg/s de capacidad, carga útil del micro-batch 600.000 × 400 B = 240 MB, ρ = 0,6, estable. Dos hechos de apoyo para el runbook operativo: el rol de IAM necesita kafka-cluster:Connect y kafka-cluster:ReadData, y el puerto debe coincidir con el modo (9092 PLAINTEXT, 9094 SASL/SSL, 9098 IAM), más VPC peering o PrivateLink entre la VPC de Databricks y la VPC de MSK (docs.../kafka/authentication, 2026-07-10).

6.3 JDBC y el rate source

JDBC es una extracción por lotes en esta plataforma: credenciales vía dbutils.secrets.get(scope="jdbc", key="username") (nunca en la URL), o la ruta gobernada .option("databricks.connection", "prod_pg") con el JAR del driver en un Volumen; la memoria del driver está limitada a 400 MiB, y el externalOptionsAllowList por defecto solo permite seis opciones de lectura. Como las escrituras a nivel de fila sobre la fuente no están soportadas, el patrón de upsert es siempre extracción → Delta con appends idempotentes:

El rate source (spark.readStream.format("rate"), rowsPerSecond por defecto 1, más rampUpTime y numPartitions) emite filas sintéticas de timestamp + value y está «intended for testing and benchmarking»: perfecto para verificar el checkpoint, el output mode y el cableado del sink Delta en dos minutos, antes de que entre en juego ningún permiso de S3 ni de MSK.

---

7. Sinks: output modes, idempotencia y MERGE

7.1 Output modes y la matriz de Delta

«Delta Lake, which backs all Unity Catalog managed tables, supports append and complete modes but not update mode» (docs .../structured-streaming/output-mode, 2026-09-11). Observa también «Joins only support the append output mode». Los appends simples de Delta son exactly-once a través del registro de transacciones: «The Delta Lake transaction log guarantees exactly-once processing, even when there are other streams or batch queries running concurrently against the table» (docs .../structured-streaming/delta-lake, 2026-08-24).

7.2 Ejemplo resuelto E6: upserts idempotentes en foreachBatch

«Debes usar foreachBatch para las operaciones merge de Delta Lake en Structured Streaming», y foreachBatch() «provides only at-least-once write guarantees», así que la idempotencia es tu trabajo (docs.../structured-streaming/foreach, fechado 2026-08-24).

Para los appends simples escritos desde foreachBatch, vincula la identidad del batch de forma explícita: .option("txnAppId", app_id).option("txnVersion", batch_id), Delta «uses txnAppId and txnVersion to identify and ignore duplicate writes». Tres reglas de las mismas páginas: si borras el checkpoint y reinicias, debes proporcionar un txnAppId distinto, porque «New checkpoints start with a batch ID of 0»; no captures ni reintentes excepciones localmente, deja que los errores se propaguen para que el orquestador reintente el batch completo y las métricas sigan siendo honestas; y prefiere una escritura de streaming por sink, ya que varios sinks dentro de foreachBatch serializan tu paralelismo.

---

8. Exactly-once de extremo a extremo

La garantía es una cadena, y cada eslabón es un componente que ya has conocido:

El contenido del checkpoint es fijo: offsets (lo que se leyó), commits (qué micro-batches llegaron al sink: «enabling exactly-once semantics»), estado y metadata (el ID único de la consulta más los ajustes de configuración). Consecuencias que deberías poder recitar:

• un checkpoint por consulta, en almacenamiento duradero, fuera de toda política de ciclo de vida;

• borrarlo o moverlo hace arrancar de cero: Kafka reproduce según startingOffsets, Auto Loader vuelve a listar (y a reprocesar el prefijo si includeExistingFiles era true en el arranque original), y las escrituras de Delta se reproducen salvo que txnAppId cambie;

• para destinos que no son Delta (una base de datos OLTP, una cola, una API HTTP), tú eres dueño de la idempotencia claveada en batchId: «Si el batchId se mantiene igual, el resultado de tu operación debe permanecer igual»;

• foreachBatch nunca se ejecuta en modo de procesamiento continuo, y en serverless solo se soportan Trigger.AvailableNow() y el obsoleto Trigger.Once().

• 📚 ¿Sabías que…? Un checkpoint almacenado junto a una tabla Delta solo sobrevive al mantenimiento si lo nombras con un guion bajo: VACUUM «omite cualquier directorio que empiece por _», y por eso la sugerencia de las propias docs es «a directory structure such as /_checkpoints» (docs .../structured-streaming/delta-lake, 2026-08-24). Es una pequeña convención de nombrado la que separa el estado de tu stream de un comando de mantenimiento rutinario.

---

9. Unificación de batch + streaming

Una ruta de código, dos triggers. El mismo lector bronze corre como un job incremental nocturno o como un stream de baja latencia; solo cambian trigger, checkpoint y cluster.

Trigger.AvailableNow «processes all files that arrived before the query start time», que es exactamente la semántica que quiere un batch programado, y los usuarios de SQL obtienen lo mismo a través de STREAM read_files(...), que a su vez usa Auto Loader para las tablas de streaming (docs .../read_files, 2026-08-03).

9.1 Ejemplo resuelto E7: lo que cuesta el job bronze nocturno

Una ejecución AvailableNow por noche, 25 minutos por ejecución, 30 ejecuciones/mes → 12,5 horas-cluster/mes en un job cluster de 4 nodos (1 driver + 3 workers). Para las emisiones usamos el nodo ilustrativo de la propia Databricks: «an instance type which runs at 3 DBU/hour … a four node jobs cluster would be charged at $2.40 ($0.2 × 3 × 4)» (blog de mejores prácticas de coste, 2022-10-18) → 12 DBU/h. Tarifas: Jobs Compute $0.15/DBU y All-Purpose $0.55/DBU (AWS, US, lista Premium, a oct 2026: fuente documental: el registro de proveedor Databricks G-Cloud 15 capturado el 2026-09-25, corroborado por lecturas independientes en 2026-10; revisa siempre system.billing.list_prices en tu propia cuenta). Lado AWS: m5.xlarge a $0.192/h (lista de precios de EC2, leída 2026-10) × 4 nodos.

La versión all-purpose cuesta 3,7× la línea de DBU, en línea con la propia regla empírica de Databricks de que los pipelines pertenecen al cómputo de jobs, y además factura el tiempo ocioso que tendrías que gestionar tú. Los workers en spot recortan la columna de AWS hasta un 90% (el driver se queda on-demand, con aviso de expulsión de dos minutos); la columna de DBU no cambia, porque el spot altera el precio de la instancia, nunca la tarifa de emisión.

• 📚 ¿Sabías que…? La vieja advertencia de que «el retraso de listado tras la escritura en S3 puede hacer que Auto Loader se salte archivos» es historia: desde el 2020-12-01 «todas las operaciones GET, PUT y LIST de S3 … ahora son fuertemente consistentes» (AWS News Blog, 2020-12-01), confirmado en la guía de usuario de S3 (consultada 2026-10). El riesgo de salto actual es la entrega de eventos, los proveedores de nube «no garantizan la entrega del 100 % de todos los eventos de archivos», y para eso existen cloudFiles.backfillInterval (modo clásico) y el backfill automático de ~24 horas (file events gestionados). Si un tutorial de 2021 te dice que construyas una solución de compromiso para la consistencia eventual de S3, puedes borrar ese párrafo.

---

10. Las trampas: doce confusiones que esta lección no te deja conservar

1. «startingOffsets=latest está bien: es el por defecto.» En una consulta nueva se salta el backlog en silencio; las consultas reanudadas usan el checkpoint de todos modos. Pon earliest en la primera ejecución y después confía en el checkpoint para siempre.
2. Borrar un checkpoint para arreglar el retraso. Arranca de cero: Kafka reproduce o se salta según startingOffsets, Auto Loader vuelve a listar, y Delta reproduce salvo que txnAppId cambie (los checkpoints nuevos reinician en el batch ID 0).
3. Estado en la ruta de datos o bajo una regla de ciclo de vida. Esquema/checkpoint dentro del prefijo de aterrizaje → autodescubrimiento; caducidad del ciclo de vida → «the stream state is corrupted».
4. «schemaLocation y checkpointLocation deben diferir siempre.» Falso: las docs permiten que sean iguales. Lo que debe diferir es estado frente a datos, y un checkpoint por carga de trabajo.
5. Delta + outputMode("update"). No soportado; usa MERGE en foreachBatch. Delta te da solo append y complete.
6. Confiar en foreachBatch para el exactly-once. Es at-least-once: vincula txnAppId/txnVersion, mantén el MERGE idempotente y nunca tragues excepciones localmente.
7. Trigger continuo + modo de listado de directorios. La explosión de coste clásica: listar el árbol entero en cada trigger. Cambia a AvailableNow o a file events.
8. Fijar maxFilesPerTrigger en DBR 18.0+. Ahí se configura de forma dinámica, y fijarlo a mano pelea contra el auto-ajuste; en runtimes antiguos, el valor por defecto de 1.000 archivos puede ser mucho demasiado pequeño para un backfill de varios TB.
9. maxFileAge significa una sola cosa. El file source lo usa como filtro de descubrimiento de 7 días; Auto Loader lo usa como estado de seguimiento de eventos con un mínimo de 14 días y una recomendación de 90 días, y en la primera ejecución acota hasta dónde ingeres.
10. cleanSource con dos lectores. El segundo lector no encuentra nada. Además: un destino de MOVE dentro de la ruta de origen causa reingesta infinita, y los movimientos entre buckets fallan con error.
11. Clases de Kafka sin shaded o el prefijo kafka. ausente. Usa kafkashaded. / shadedmskiam., y cuando una conexión la gobierne un service credential de UC, no fijes tú kafka.security.protocol.
12. «maxOffsetsPerTrigger se restablece en cada reinicio.» Los controles de admisión se aplican a consultas nuevas; una consulta reanudada lee su checkpoint. Dimensionalo una vez, con intención, y vigila el retraso en lugar de re-ajustarlo tras cada incidente.

Veredicto comparativo: listado de directorios frente a notificaciones de archivos (a oct 2026)

• Elige los file events gestionados (una cola por external location de Unity Catalog, cloudFiles.useManagedFileEvents=true, DBR 14.3 LTS+) para la mayoría de las cargas de trabajo de producción: esa es la recomendación explícita de Databricks: «Databricks recommends file notification mode using file events for most workloads» (docs.../auto-loader/file-notification-mode, 2026-10-07). Obtienes descubrimiento dirigido por eventos sin LIST por trigger (E3: $0.30/mes frente a $28.80 to $25,920/mes en el mismo prefijo de 2M de archivos), permisos de una vez en lugar de IAM por stream, backfills completos automáticos al activarlos y unos cada 24 horas, y el requisito duro de ejecutar el stream al menos una vez cada 7 días para que la caché de file events no caduqué.
• Quédate en el listado de directorios cuando el prefijo sea pequeño y de corta vida, cuando el job sea una ejecución programada de Trigger.AvailableNow (96 listados/día ≈ $28.80/mes con 2M de archivos: un error de redondeo al lado del cómputo), cuando no puedas conceder s3:GetBucketNotification / s3:PutBucketNotification más los derechos de SNS/SQS, o cuando necesites cambiar la ruta de entrada sin un checkpoint nuevo (una capacidad exclusiva del listado de directorios). Es el por defecto precisamente porque no necesita nada más allá de acceso de lectura.
• Las notificaciones de archivos clásicas SNS+SQS son el término medio heredado: dirigidas por eventos como los file events, pero con el conjunto de acciones IAM publicado, un prefijo de recurso que gobernar (databricks-auto-ingest-*), un límite duro de 100 pipelines por bucket de S3, colas por stream que limpiar, entrega best-effort que exige cloudFiles.backfillInterval y ninguna movilidad de la ruta de entrada. Los diseños nuevos pertenecen a los file events gestionados; el modo clásico sobrevive en entornos anteriores a las external locations de Unity Catalog.
• La respuesta-trampa: «las notificaciones de archivos cuestan más porque pagas por SNS y SQS.» La factura que de verdad puedes modelar, el LIST de S3 a $0.005 por 1.000, escala con la frecuencia de listado, y las fuentes de esta lección no traen ningún precio verificado de SQS/SNS (así que no lo modeles de memoria). Decide sobre permisos, comportamiento de backfill, la regla de la caché de 7 días y el límite de 100 por bucket; trata los céntimos de solicitudes como secundarios.

---

Casos Reales

📋 Caso A: Asurion (seguros y protección de dispositivos, cloud: AWS) El reto. Un entorno heredado de 8.000 tablas, 10.000 vistas y ~600 data marts, con mapeos ETL por tabla y una capa batch/speed duplicada que no podía absorber nuevas fuentes de forma barata. Stack en AWS. Spark Structured Streaming, Auto Loader y foreachBatch hacia tablas Delta, ingiriendo «database CDC feeds, APIs and flat files to streams from Kinesis, Kafka, SNS and SQS», con Databricks SQL alimentando 600 data marts: un data lake de producción explícitamente «on AWS» (https://www.databricks.com/blog/2021/09/16/large-scale-etl-and-lakehouse-implementation-at-asurion.html, 2021-09-16). Resultado reportado. «The Asurion data lake executes 4,000+ streaming jobs and hosts over 4,000 tables in a production data lake on AWS» (https://www.databricks.com/blog/2021/05/10/aws-guide-to-data-ai-summit-featuring-disney-comcast-capital-one-and-mcdonalds.html, 2021-05-10). En arquitectura: «an unified approach to data movement from traditional ETL to streaming data pipelines between Delta tables» (2021-09-16). Cómo encaja en esta lección. Secciones 1, 6 y 7: con más de 4.000 streaming jobs, la arquitectura es un framework de ingesta parametrizado (un patrón Auto Loader / foreachBatch, muchos feeds), y la lista de fuentes nativas de AWS (Kinesis, SNS, SQS) es exactamente por qué importan las decisiones de sustrato de las secciones 4 y 6. ⏳ Nota temporal: cifras de 2021 reportadas en su momento; revisa cualquier afirmación sobre concurrencia de workspace de ese blog contra la documentación actual antes de reutilizarla.

📋 Caso B: Burberry (venta minorista de lujo, cloud: AWS) El reto. Lotes diarios de clickstream que llegaban «anywhere between 2:00 and 7:00 PM GMT», y un almacén en la nube que preagregaba el detalle que Customer 360 necesitaba. Stack en AWS. Databricks Data Intelligence Platform con Snowplow Behavioral Data Platform para Customer 360 en tiempo real: la página del caso lleva la propia etiqueta Cloud: AWS de Databricks (https://www.databricks.com/customers/burberry, página sin fecha, consultada 2026-10). Resultado reportado. «99% Reduction in latency for customer clickstream data · 52x Increase in cookie duration»; «we've broken free from daily data batches and formed an accurate real-time view of our customers' digital activity» (misma URL, consultada 2026-10; origen del caso 2023). Cómo encaja en esta lección. Secciones 4 y 9: un SLA de latencia sobre archivos de evento es un requisito de negocio, y se cumple con descubrimiento dirigido por eventos más un trigger de baja latencia, no con un cluster más grande. El cuello de botella era la cadencia de llegada de archivos, no el cómputo. ⏳ Nota temporal: una cifra con origen en 2023 que aún se muestra en 2026: cítala como lo que Burberry reportó en su momento.

---

Actualizaciones de 2026 (en octubre de 2026)

---

---

Preguntas de Práctica (Practice Questions)