El job de ventas de la lección anterior funcionaba, pero cada pregunta nueva costaba un job más: ventas por productor y mercado era uno, el ranking por mercado otro, unirlo con el catálogo un tercero, y entre cada uno la salida completa iba a HDFS y volvía. Apache Spark nació en Berkeley (2009, Matei Zaharia) precisamente de esa frustración: los algoritmos iterativos de aprendizaje automático y las consultas interactivas tardaban en MapReduce diez o cien veces más de lo que la CPU justificaba, porque el tiempo se iba en escribir y leer disco entre fases. Spark conserva lo esencial del modelo (datos en particiones, tareas independientes y reejecutables, un shuffle para agrupar por clave) pero expresa todo el cálculo como un DAG de operadores que un planificador convierte en fases y que se ejecuta manteniendo los datos intermedios en memoria. Sobre esa base construyó una API de colecciones (RDD), después una de tablas con optimizador (DataFrames y Spark SQL), y encima librerías de aprendizaje automático (MLlib), grafos (GraphX) y flujos (Structured Streaming, que veremos en 05-04). Hoy es el motor de lotes de referencia. En esta lección analitica reescribe el cálculo de ventas diarias como servicios/analitica/ventas_diarias.py, con DataFrames y con RDD para comparar, lo lanza en local y en un clúster de docker-compose.yml, lee el plan de ejecución, y entrena unas recomendaciones con ALS sobre los clics.

1. Por qué Spark: un DAG en memoria frente a fases en disco
2. Arquitectura: driver, cluster manager, executors, tasks y stages
3. RDD: colecciones inmutables, transformaciones perezosas y linaje
4. DataFrames y Spark SQL: Catalyst y formatos columnares
5. Operaciones estrechas y anchas, shuffle y stages
6. Optimizaciones: cache, broadcast join, particionado y sesgo
7. MapReduce frente a Spark
8. MLlib: recomendaciones con ALS sobre los clics
9. Práctica: servicios/analitica/ventas_diarias.py
10. Errores Comunes y Consejos
11. Ejercicios
12. Conclusión

En MapReduce (05-02), una cadena de tres agrupaciones son tres jobs y seis pasos por disco: cada job escribe su salida en HDFS (replicada tres veces) y el siguiente la lee. Para un algoritmo iterativo que recorre los mismos datos cien veces, son cien lecturas completas desde disco de un conjunto que no ha cambiado. Spark cambia dos cosas:

• Todo el cálculo es un único programa, un DAG de operadores (leer, filtrar, explotar, agrupar, unir, escribir) que el motor conoce completo antes de ejecutar nada. Con esa visión global puede encadenar en una sola pasada los operadores que no necesitan redistribuir datos, y solo materializa datos intermedios en los puntos donde hay un shuffle. Es el patrón dataflow de 05-01.
• Los datos intermedios viven en memoria de los ejecutores, y se pueden cachear explícitamente para reutilizarlos en varias pasadas. Cien iteraciones sobre un conjunto cacheado leen disco una vez.

El precio de no escribir a disco es la tolerancia a fallos: si un nodo muere, sus datos intermedios en memoria desaparecen. La solución de Spark, y su idea original, es el linaje (apartado 3): en lugar de guardar los datos intermedios, guarda la receta para recalcularlos a partir de la entrada, y recomputa solo las particiones perdidas. Es la reejecución determinista de 05-01, aplicada a fragmentos de un cálculo en vez de a tareas enteras.

• Driver. El proceso que ejecuta el programa principal (ventas_diarias.py). Contiene el SparkSession (antes SparkContext), construye el DAG, lo divide en jobs, stages y tasks, las envía a los executors y recoge resultados. Es el maestro de 05-02, uno por aplicación: si el driver muere, la aplicación muere (en YARN en cluster mode, el driver es el ApplicationMaster y se puede reintentar).
• Cluster manager. Reparte recursos entre aplicaciones. Spark trae el suyo (standalone, un maestro y workers: el que usaremos en docker-compose.yml), y se integra con YARN (05-02: el driver pide contenedores al ResourceManager) y con Kubernetes (07-05: cada executor es un pod). Al driver le da igual cuál sea; la API es la misma.
• Executor. Un proceso JVM en un nodo worker, con N núcleos y M GB asignados, que vive toda la aplicación. Ejecuta tasks en hilos (un task por núcleo a la vez) y guarda en memoria las particiones cacheadas y los datos de shuffle. A diferencia de MapReduce, no se arranca una JVM por tarea: la JVM se arranca una vez y ejecuta miles de tareas, lo que elimina el coste fijo por tarea.
• Task. La unidad de trabajo: ejecutar una cadena de operadores sobre una partición. Una fase con 200 particiones son 200 tasks.
• Stage. Un conjunto de tasks que se pueden ejecutar sin shuffle. El DAG se corta en stages en cada operación ancha (apartado 5); dentro de un stage, los operadores se encadenan (pipelining) y una fila pasa por todos ellos sin tocar disco.
• Job. Todo lo que desencadena una acción (apartado 3): write, collect, count. Un programa puede tener varios jobs; cada job tiene uno o más stages.

Con PySpark, el driver es un proceso Python que se comunica con una JVM (vía Py4J), y en los executors el código Python de las funciones de RDD se ejecuta en procesos Python auxiliares con serialización de ida y vuelta; con DataFrames, en cambio, la mayoría de operaciones se traducen a código JVM y Python no interviene por fila. Es la razón principal por la que la API de DataFrames es la recomendada.

El RDD (Resilient Distributed Dataset) es la abstracción original de Spark: una colección de elementos particionada entre los executors, inmutable (no se modifica: se crea otro RDD a partir de él) y resiliente por linaje. Se opera con dos tipos de métodos:

La pereza es la que permite optimizar: cuando el programa dice filter y luego map y luego count, Spark no ejecuta tres pasadas; al llegar a count conoce toda la cadena y la ejecuta en una sola pasada por partición. Y el linaje es el DAG de transformaciones que llevó a cada RDD: si se pierde una partición del RDD ventas porque murió su executor, Spark mira el linaje (ventas = lineas.reduceByKey(...), lineas = eventos.flatMap(...), eventos = sc.textFile(...)) y recalcula solo esa partición desde el bloque de HDFS correspondiente. No hay que guardar nada intermedio en disco; el coste es recomputar, que es asumible mientras el linaje sea corto (para linajes largos, iterativos, existe checkpoint(), que sí materializa en HDFS y corta el linaje).

El cálculo de ventas con RDD, que servirá de comparación en la práctica:

Hasta aquí no se ha leído ni un byte: textFile, map, filter, flatMap y reduceByKey son transformaciones. ventas.collect() o ventas.saveAsTextFile(...) lanzarían el job. Dos detalles que distinguen a un usuario de Spark: reduceByKey hace la reducción local en cada partición antes del shuffle (el combiner de 05-02, automático), mientras que groupByKey seguido de una suma mueve todos los valores por la red; y las funciones lambda viajan serializadas a los executors, así que no pueden capturar objetos no serializables (una conexión a base de datos, por ejemplo).

Un RDD es una colección de objetos opacos para Spark: no sabe que ln["productor"] es una columna, así que no puede optimizar nada más allá del encadenamiento. Un DataFrame es una tabla con esquema (columnas con nombre y tipo), distribuida en particiones como un RDD, y sobre la que se opera con una API relacional (select, filter, groupBy, join, agg) o directamente con SQL (spark.sql("SELECT ...")). Ambas producen el mismo plan lógico, que pasa por Catalyst, el optimizador:

1. Análisis: resolver nombres de columnas y tipos contra el esquema.
2. Optimización lógica: reglas como empujar los filtros hacia la lectura (predicate pushdown: filtrar tipo = 'pedido.creado' al leer, no después), leer solo las columnas necesarias (column pruning), simplificar expresiones, reordenar joins.
3. Planificación física: elegir algoritmos: un join se hace por broadcast hash si un lado es pequeño, por sort-merge si no; una agregación en dos fases (parcial en cada partición, final tras el shuffle).
4. Generación de código: Tungsten compila el plan a bytecode Java especializado (whole-stage codegen) y gestiona la memoria fuera del heap de la JVM en formato binario compacto. Es lo que hace que PySpark con DataFrames sea tan rápido como Scala: Python solo describe el plan.

Desde Spark 3, el Adaptive Query Execution (AQE) reoptimiza el plan durante la ejecución con estadísticas reales: reduce el número de particiones tras un shuffle pequeño, convierte un sort-merge join en broadcast si descubre que un lado cabe, y divide particiones sesgadas (apartado 6).

Los DataFrames se combinan con los formatos columnares. Un fichero Parquet guarda los datos por columnas en lugar de por filas: todos los productor juntos, todos los importe juntos, comprimidos con codificaciones adecuadas a cada tipo (diccionario para cadenas repetidas, run-length para valores consecutivos) y con estadísticas (mínimo, máximo, nulos) por bloque de filas. Una consulta que solo necesita productor e importe lee esas dos columnas y salta el resto del fichero, y un filtro dia = '2026-09-14' salta bloques enteros cuyo rango no lo contiene. Frente a JSONL, que obliga a leer y parsear cada byte, Parquet reduce en un orden de magnitud tanto los bytes leídos como el tamaño en disco. Por eso el lago de 04-02 empieza en JSONL (lo que produce Kafka) y el primer paso de la analítica lo convierte a Parquet, particionado por día.

La división del DAG en stages depende de un único criterio: qué dependencia tiene cada partición de salida respecto a las de entrada.

El DAG de ventas tiene exactamente un shuffle, el groupBy, y por tanto dos stages. Todo lo demás se encadena: una fila del JSON se filtra, se explota y se agrega parcialmente en la misma task sin que nadie la escriba. El join con el catálogo, que en MapReduce sería un tercer job, es estrecho gracias al broadcast (apartado 6). En la interfaz web de Spark (http://localhost:4040 durante la ejecución) cada job aparece con sus stages, cada stage con sus tasks, y para cada stage los bytes de shuffle write y shuffle read: son los contadores de 05-02, y el criterio de diagnóstico es el mismo.

El shuffle de Spark escribe también en disco local (los ficheros de shuffle, servidos por los executors o por un external shuffle service), así que "en memoria" no significa "sin disco": significa que entre operaciones estrechas no se escribe, y que los datos cacheados se sirven de memoria. Con 200 particiones de shuffle por defecto (spark.sql.shuffle.partitions), un job pequeño produce 200 tasks minúsculas en el segundo stage; AQE las fusiona, pero en versiones antiguas o con AQE desactivado conviene bajar el número.

cache() y persist(). Marcan un DataFrame o RDD para que, la primera vez que se calcule, sus particiones se guarden (en memoria por defecto; persist(StorageLevel.MEMORY_AND_DISK) permite desbordar a disco, DISK_ONLY, o serializado para ahorrar memoria). Compensa cuando el mismo resultado intermedio se usa en varias acciones: el DataFrame lineas de la práctica alimenta las ventas por productor, un ranking y una comprobación de calidad; sin cache, cada acción vuelve a leer y parsear el JSON de HDFS. Con unpersist() se libera. Cachear algo que se usa una vez es puro coste.

Broadcast join. Unir las ventas (millones de filas, repartidas) con el catálogo (unas decenas de filas) no debería mover las ventas. Con F.broadcast(catalogo), Spark envía una copia del catálogo a cada executor y el join se resuelve localmente, sin shuffle del lado grande: es el map-side join de 05-02, automático. Spark lo hace solo si estima que el lado pequeño ocupa menos de spark.sql.autoBroadcastJoinThreshold (10 MB por defecto); forzarlo con broadcast() conviene cuando la estimación falla (un CSV sin estadísticas). Con dos lados grandes, el plan es un sort-merge join: shuffle de ambos por la clave y mezcla ordenada.

repartition(n) y coalesce(n). El número de particiones gobierna el paralelismo. repartition(n) hace un shuffle completo para obtener n particiones equilibradas (o repartition("dia") para colocar cada día en su partición, útil antes de escribir particionado); coalesce(n) reduce el número sin shuffle, fusionando particiones locales, y es la forma barata de no escribir 200 ficheros Parquet de 3 KB. Como orientación: particiones de 100–200 MB en memoria y entre 2 y 4 tasks por núcleo disponible.

Sesgo con salting. La Semana del Queso Artesano vuelve: al agrupar por productor, la partición de queseria-montblanc recibe la mitad de las filas. En DataFrames, la agregación parcial por partición (que Catalyst inserta siempre) alivia el caso de las sumas, igual que el combiner; el sesgo duele cuando la operación no se prerreduce (un join sort-merge, collect_list, ventanas). La técnica de 05-01, ahora con columnas:

Y desde Spark 3, AQE con spark.sql.adaptive.skewJoin.enabled=true detecta particiones de join sesgadas (por tamaño relativo a la mediana) y las divide automáticamente, sin sal manual. Para agregaciones sesgadas sin prerreducción posible sigue haciendo falta la sal.

La ventaja de Spark en el job de ventas de 05-02 se ve en los números: el mismo cálculo que en MapReduce tardaba 52 s en el clúster de pruebas tarda 6 s en Spark sobre YARN con los mismos recursos, y encadenar el ranking y el join con el catálogo no añade jobs, sino un stage. La contrapartida es la memoria: un executor mal dimensionado (demasiado poca memoria para el shuffle o la cache) falla con OutOfMemoryError donde MapReduce simplemente escribía a disco.

El segundo caso de la analítica de Kilómetro Cero, "productos recomendados para Ana", es un problema iterativo: exactamente el tipo de problema que MapReduce resolvía mal. MLlib trae algoritmos distribuidos listos, y para recomendación el clásico es ALS (Alternating Least Squares), una factorización de la matriz clientes × productos que aprende un vector de factores por cliente y otro por producto a partir de interacciones (compras, clics) y predice el interés de cada pareja no observada. Con clics no hay "puntuación", sino señales implícitas (vio, añadió al carrito, compró), y ALS tiene un modo para eso.

Los clics del lago, /km0/clics/2026-09-14/hora=13/web-01.jsonl (04-02), tienen esta forma:

Cada iteración de fit es un job con varios stages sobre los mismos datos, que MLlib cachea internamente; con 10 iteraciones en MapReduce serían 20 jobs y 20 lecturas de HDFS. El resultado (ana → vino-crianza, lucia → queso-curado...) lo carga el pipeline de 05-05 en la base de catalogo para que la web lo muestre. Lo que MLlib no decide es la calidad: elegir rank, regParam y los pesos de las acciones exige evaluar (RegressionEvaluator o métricas de ranking sobre un conjunto reservado), y eso pertenece a un curso de aprendizaje automático, no a este.

Al docker-compose.yml de 04-02 (HDFS) y 05-02 (YARN) se añade un clúster Spark standalone, que es más ligero que YARN para desarrollar. El driver correrá en nuestra máquina o en el contenedor del maestro:

Los dos workers ofrecen 4 núcleos en total, y HDFS es alcanzable como hdfs://namenode:8020 desde la misma red de Compose. El catálogo es un CSV pequeño, servicios/analitica/catalogo.csv, que en producción vendría de una exportación diaria de km0_catalogo:

Puntos clave del programa:

• Esquema explícito. Sin él, spark.read.json hace una pasada completa solo para inferir tipos. Con él, la lectura es una pasada y los tipos son los que queremos (fecha_ms como LongType, no double).
• explode convierte el array lineas en filas, que es lo que hacía el bucle for ln in ev["datos"]["lineas"] del mapper.
• Un solo shuffle. El groupBy es la única operación ancha. filter, select, explode y el join broadcast se encadenan en el mismo stage que la lectura. Catalyst inserta una agregación parcial antes del shuffle (HashAggregate con partial_sum), así que lo que cruza la red son parciales por (dia, mercado, productor) por partición: el combiner, sin escribirlo.
• cache() en lineas porque hay dos acciones (control.first() y el write); sin él, el JSON se leería y parsearía dos veces.
• partitionBy("dia") con partitionOverwriteMode=dynamic. La salida es un directorio por día (dia=2026-09-14/), y overwrite en modo dinámico reemplaza solo los días que este job escribe, dejando intactos los demás. Ejecutar dos veces el job del 14 de septiembre produce exactamente la misma salida: es la idempotencia por partición que el pipeline de 05-05 necesita para reprocesar y hacer backfill.
• coalesce(1) antes de escribir, porque el agregado son unas 12 filas por día y no queremos 8 ficheros Parquet de 2 KB. Con agregados grandes se quitaría o se ajustaría.

En local, con cuatro hilos como executors simulados (local[4]), sobre el fichero generado en 05-01:

El plan se lee de abajo arriba, y cada línea confirma una decisión de los apartados anteriores: FileScan json con PushedFilters (el filtro empujado a la lectura), Generate explode, InMemoryRelation (la cache), el HashAggregate con partial_sum antes del Exchange hashpartitioning(..., 8) (agregación parcial, después el único shuffle con 8 particiones) y el HashAggregate final, y BroadcastHashJoin con BroadcastExchange del CSV (el catálogo viaja, las ventas no). Si en lugar de BroadcastHashJoin apareciera SortMergeJoin con dos Exchange, sabríamos que el broadcast no se aplicó y que estamos pagando un shuffle de más.

Contra el clúster standalone de Compose, con el fichero en HDFS:

Con --master yarn y la configuración de Hadoop en HADOOP_CONF_DIR, el mismo fichero se lanzaría sobre el YARN de 05-02, con el driver como ApplicationMaster (--deploy-mode cluster), sin cambiar una línea de código. En la interfaz del maestro (localhost:8080) se ven los workers y las aplicaciones; en la del driver (localhost:4040, mientras corre) los jobs, stages y tasks, con sus bytes de shuffle.

ventas_rdd.py contiene la función del apartado 3. Lanzado con --rdd, produce las mismas cifras, y en la interfaz se ven dos diferencias: el stage de lectura es más lento (cada línea pasa por un proceso Python con json.loads, en lugar del parser JSON de la JVM) y no hay plan que leer: Spark ejecuta las lambdas tal cual, sin empujar filtros ni podar columnas, porque no sabe qué hacen. En el fichero de 130 MB la diferencia es de 9 s frente a 4 s en local[4]; en terabytes, de horas. El RDD sigue siendo la herramienta correcta cuando los datos no tienen esquema o la lógica no cabe en expresiones de columna, y es lo que hay debajo de todo DataFrame; pero para analítica sobre eventos con esquema, DataFrames es la API.

• collect() sobre un DataFrame grande. Trae todas las filas al driver, que tiene unos GB de memoria: OutOfMemoryError en el driver. Para mirar, show() o take(n); para guardar, write.
• Olvidar que las transformaciones son perezosas. Un filter mal escrito no falla al escribirlo, sino en la primera acción, con una traza que apunta al write. Y un print dentro de una lambda no aparece en el driver: se ejecuta en los executors (mira sus logs).
• groupByKey + suma en RDD. Mueve todos los valores por la red. reduceByKey o aggregateByKey combinan localmente. En DataFrames, groupBy().agg() ya lo hace.
• UDF de Python por fila. Una udf en PySpark serializa cada fila hacia un proceso Python y de vuelta; anula Catalyst y Tungsten. Casi todo se puede expresar con F.*; si no, usar pandas_udf (vectorizada por lotes con Arrow).
• Cachear sin usar o no liberar. cache() sobre algo que se usa una vez es un coste sin beneficio; cachear muchas cosas sin unpersist() desborda la memoria de los executors y provoca desalojos y recomputaciones silenciosas.
• 200 particiones de shuffle para 10 MB. El valor por defecto de spark.sql.shuffle.partitions es para clústeres grandes. Ajústalo o confía en AQE; y coalesce antes de escribir para no producir cientos de ficheros de un kilobyte que ahogarán el NameNode (04-02).
• Confiar en el broadcast automático con un CSV. Sin estadísticas, Spark puede no estimar el tamaño y hacer un sort-merge join. F.broadcast() explícito y comprobar en explain().
• Inferir el esquema del JSON en producción. Es una pasada extra y un esquema que cambia con los datos (un día sin version y la columna desaparece). Esquema explícito, versionado con el contrato del evento (02-05).
• Escribir con overwrite sin partitionOverwriteMode=dynamic. Borra todo el directorio de salida, incluidos los días que no se estaban recalculando. Es el error más caro que se comete con un pipeline de backfill.

Amplía ventas_dataframe para que, además de las ventas, produzca para cada (dia, mercado) el productor con más ventas y su cuota sobre el total del mercado (porcentaje), usando funciones de ventana (pyspark.sql.Window). ¿Cuántos shuffles añade tu solución? Compáralo con los dos jobs del ejercicio 1 de 05-02.

Este es el plan de una versión modificada del programa, escrita por un compañero. Identifica cuatro problemas de rendimiento a partir del plan y di cómo corregir cada uno.

Los eventos del 8 al 14 de septiembre de 2026 (la Semana de la Vendimia) tenían un error en el precio de vino-crianza que pedidos ha corregido regenerando los ficheros pedidos.jsonl de esos siete días en HDFS. Escribe la invocación (o las invocaciones) de ventas_diarias.py que recalcule exactamente esos siete días sin tocar los demás, y explica qué combinación de opciones del programa garantiza que (a) no queden datos antiguos de esos días, (b) no se borren los otros días, y (c) ejecutarlo dos veces dé el mismo resultado. ¿Qué cambiarías si la entrada fuera un glob pedidos-*.jsonl con varios ficheros por día?

Ejercicio 1.

Las dos ventanas comparten la partición (dia, mercado), así que Spark añade un shuffle (Exchange hashpartitioning(dia, mercado)) y un Sort dentro de cada partición para row_number; con AQE y las mismas claves, a veces reutiliza el intercambio. Total: dos shuffles en el programa (el groupBy y la ventana), en un solo job con tres stages, frente a los dos jobs de MapReduce con sus cuatro pasos por HDFS. Y el join con el catálogo sigue sin costar shuffle porque ventas ya lo traía resuelto.

Ejercicio 2.

1. SortMergeJoin con dos Exchange por productor en lugar de BroadcastHashJoin: el catálogo de tres filas está provocando un shuffle de las ventas y una ordenación. Corrección: F.broadcast(catalogo).
2. BatchEvalPython [calcular_importe]: una UDF de Python por fila, que además impide que el partial_sum se compute en la JVM sin salir a Python. Corrección: F.col("ln.cantidad") * F.col("ln.precio").
3. Exchange ... 200: 200 particiones de shuffle para un agregado de decenas de filas; 200 tasks minúsculas por stage. Corrección: spark.sql.shuffle.partitions a 8 (o AQE con coalescencia activada).
4. FileScan json [id_evento, tipo, version, fecha_ms, origen, datos] sin PushedFilters ni poda: se leen todas las columnas, y probablemente sin esquema explícito (inferencia, una pasada extra). Corrección: esquema declarado y select de las columnas necesarias justo tras la lectura; el filtro tipo debería aparecer como PushedFilters (lo hace cuando la columna se compara con un literal y el esquema es conocido).

Un quinto detalle: no hay InMemoryRelation, así que si el programa hace más de una acción, releerá el JSON cada vez.

Ejercicio 3.

Una sola invocación con un glob para los siete días, o siete invocaciones (una por día, que es lo que hará Airflow en 05-05 con el backfill):

(a) mode("overwrite") reemplaza el contenido de cada partición dia=2026-09-NN/ que el job escribe, de forma atómica por partición (escritura en temporal y commit). (b) partitionOverwriteMode=dynamic limita el borrado a las particiones presentes en la salida del job; sin él, overwrite vaciaría ventas_diarias/ entero, incluidos agosto y el resto de septiembre. (c) El cálculo es determinista (misma entrada, mismo agregado) y la escritura reemplaza la partición completa: dos ejecuciones dejan los mismos bytes (salvo nombres de fichero internos), sin duplicar ni acumular. Es importante que dia se derive de fecha_ms (tiempo de evento) y no del nombre del directorio: si un evento del 14 llegara tarde y pedidos lo hubiera dejado en el fichero del 15, el job del 15 lo escribiría en dia=2026-09-14/, sobrescribiendo la partición del 14 con solo ese evento. Para evitarlo hay dos opciones: filtrar en el job por el rango de días que se está procesando (F.col("dia").between(...)) y descartar o registrar los fuera de rango, o que la partición de entrada y salida coincidan por construcción. Con un glob pedidos-*.jsonl por día no cambia nada en el programa (spark.read.json acepta globs y directorios); solo conviene comprobar que ninguno de los ficheros esté aún en escritura, que es exactamente lo que el sensor de 05-05 vigilará con _SUCCESS o con un fichero de cierre.

Spark toma el modelo de MapReduce (particiones, tareas reejecutables, shuffle por clave) y le quita lo que lo hacía lento: el cálculo entero es un DAG de operadores que el driver conoce completo, corta en stages solo donde hay una dependencia ancha, encadena todo lo demás en la misma task, y mantiene los datos intermedios en la memoria de executors que viven toda la aplicación. La tolerancia a fallos viene del linaje, que recomputa la partición perdida en lugar de haberla escrito a disco. Sobre los RDD, inmutables y perezosos, los DataFrames añaden esquema y un optimizador, Catalyst, que empuja filtros, poda columnas, inserta agregaciones parciales y elige broadcast o sort-merge para cada join; Tungsten compila el plan, y Parquet hace que leer dos columnas de un año de eventos cueste lo que ocupan esas dos columnas. Hemos aprendido a leer un explain() para verificar que el plan hace lo que creemos, a usar cache cuando hay varias acciones, broadcast para el catálogo, coalesce antes de escribir, la sal para Quesería Montblanc, y partitionBy con sobreescritura dinámica para que el job de ventas diarias sea idempotente por día. servicios/analitica/ventas_diarias.py calcula ahora en seis segundos, y en un solo programa, lo que en 05-02 eran tres jobs y un minuto; y ALS en MLlib ha convertido los clics del lago en recomendaciones sin que las cien iteraciones supongan cien lecturas.

Todo lo que hemos hecho parte de un conjunto acotado: el fichero del 14 de septiembre, ya cerrado, o los clics de siete días. Pero pedidos.eventos no se cierra nunca: los eventos stock.actualizado llegan a razón de cientos por segundo, y las posiciones de los repartidores a 2,4 millones al día, y ni el panel de reparto ni la alerta de stock bajo pueden esperar al lote de la noche. La siguiente lección trata el procesamiento de flujos: qué cambia cuando el conjunto no termina, cómo se define "los últimos cinco minutos" cuando los eventos llegan desordenados (las marcas de agua), y cómo Flink y Spark Structured Streaming ejecutan el mismo DAG de esta lección sobre un flujo que no se detiene.