Día 2 -- Features de mantenimiento predictivo con PySpark Streaming

## Martes, 9:00 AM -- La reunión con el equipo de ML

Patricia ha convocado una reunion a primera hora en la sala de control de planta -- una sala con 6 monitores mostrando dashboards de Grafana en tiempo real. Ademas de ti, están Aitor Etxebarria (ML Engineer senior, 30 años, doctorado en predictive maintenance por Tecnun) y Miren Zubeldia (ML Engineer, 28 años, ex-Siemens Digital Industries). El tema: el modelo de mantenimiento predictivo no funciona tan bien como esperaban.

1Patricia: "El modelo actual predice fallo de rodamiento con 72% de 
2precisión. Los operarios no se fian -- si 3 de cada 10 alertas son 
3falsas alarmas, dejan de hacer caso. Necesitamos al menos 90% para 
4que confien y actuen sin cuestionar."
5
6Aitor: "El problema no es el modelo -- es un XGBoost bastante bueno. 
7El problema son las FEATURES. Lo único que recibimos del pipeline 
8actual es media_vibracion_1h y media_temperatura_1h por maquina. 
9Eso es como intentar diagnosticar una enfermedad solo con la 
10temperatura del paciente -- necesitas más datos.
11
12Lo que necesitamos (y hemos validado offline con datos históricos):
13
141. Percentil 95 de vibracion en ventana de 1 hora
15   - Los picos importan MAS que la media. Un rodamiento degradado
16     genera picos esporadicos antes de fallar consistentemente.
17
182. Derivada de temperatura (cambio por minuto, no valor absoluto)
19   - Una maquina a 80C estable es normal. Una que pasa de 60 a 85 
20     en 15 minutos tiene un problema -- eso es +1,67 grados por 
21     minuto, cuando una maquina estable se queda en practicamente cero.
22
233. Ratio alta/baja frecuencia de vibracion
24   - Cuando el armonico de fallo (4x husillo) supera el 30% del 
25     fundamental, hay degradación. Necesitamos ese ratio.
26
274. Correlacion cruzada temperatura-vibracion por maquina (1h)
28   - Si temperatura y vibracion suben juntas = friccion por desgaste.
29   - Si solo sube vibracion pero no temp = desbalanceo (menos grave).
30
31Con estas 4 features, en nuestras pruebas offline subimos de 72% a 94%."
32
33Fernando (que se ha asomado): "Necesito ese pipeline de features en 
34producción en 2 semanas. Que las features se calculen en streaming 
35con ventana de 1 hora y esten disponibles para el modelo en menos de 
365 minutos desde que el dato entra. Es la prioridad número 1."

Merece la pena parar un segundo en ese 72%, porque es el numero que vas a mover todo el dia. De cada diez veces que el modelo grita "esta maquina va a fallar", siete aciertan y tres son falsa alarma. Eso es lo que hace que los operarios dejen de hacer caso. Pero hay un segundo numero que nadie ha mencionado y que conviene sacar en la reunion: de cada diez averias que ocurren de verdad, cuantas aviso el modelo? Las dos cosas se mueven en sentidos contrarios con el mismo mando. Si subes el umbral de alerta, mandas menos avisos y casi todos buenos -- los operarios encantados -- pero se te cuelan averias sin avisar. Si lo bajas, avisas de todo y vuelves a las falsas alarmas. Ese segundo numero tiene nombre, recall o sensibilidad, y es el porcentaje de averias reales que el modelo llego a avisar. Asi que cuando en la reunion se pide un 90%, la pregunta de vuelta es: y cuantas averias estamos dispuestos a que se nos escapen para conseguirlo? Esa conversacion la tiene que provocar el ingeniero de datos, porque es el unico que esta viendo que el mando es uno solo.

## El reto técnico: features en streaming vs batch

Después de la reunión, vuelves a tu escritorio con Kenji. "Las features que pide Aitor son complejas", te dice. "La media es trivial en streaming, pero el percentil 95 requiere mantener un buffer de todos los valores de la ventana. La correlación cruzada requiere joins entre sensores de vibración y temperatura de la misma máquina. Y la derivada necesita estado del minuto anterior. Todo eso con 3.600 eventos de vibración por segundo más los 811 de temperatura."

Analizas el problema. Tienes dos opciones de computo:

• ▹Opción A -- Spark Structured Streaming puro: procesas cada micro-batch (trigger: 30s), calculas features con window functions y watermarks. Ventaja: stack actual, equipo lo conoce. Desventaja: percentil 95 y correlación son costosos en streaming.
• ▹Opción B -- Hybrid: Spark Streaming calcula features simples (media, max, min, conteo) y escribe a una tabla Iceberg temporal cada minuto. Un segundo job batch (cada 5 min) lee esa tabla y calcula las features complejas (p95, correlación). Ventaja: más fácil de debuggear. Desventaja: latencia 5-7 min en lugar de 2-3 min.

Decides la Opción A: streaming puro. La razón: Fernando dijo <5 minutos y Patricia idealmente quiere 2 minutos. Con hybrid pierdes 5-7 minutos y estás fuera del SLA. Además, Spark Structured Streaming soporta agregaciones sobre ventanas temporales con estado manejado -- es complejo pero factible.

Echa la cuenta de latencia entera antes de darla por buena, y ojo con de donde sale cada sumando. No es el trigger: el trigger solo dice cada cuanto despierta Spark. Con salida en modo append, una ventana no produce resultado hasta que el watermark deja atras su final, asi que la latencia es el watermark mas el paso de la ventana. Con paso de 5 minutos son hasta 7, que es justamente el rango por el que acabas de descartar la opcion B. Con paso de 1 minuto son 3, y te quedan dos de margen. Lo que pagas por bajar el paso es estado: de doce ventanas abiertas por maquina pasas a sesenta, y el estado de las agregaciones sube de unos 5 MB a unos 25. Sigue siendo poco al lado de los 100 MB del join, asi que el cambio sale barato. Lo otro que pagas son ficheros mas pequeños en el lake, y eso se compacta despues.

## Arquitectura del pipeline de features

El pipeline que vas a construir tiene esta estructura:

1. 01.Lee de Kafka topics: sensors.vibration.v2 + sensors.temperature (join por machine_id)
2. 02.Aplica watermark de 2 minutos (tolera eventos que llegan con retraso)
3. 03.Agrupa por machine_id + ventana de 1 hora deslizante con paso de 1 minuto. Ojo con el paso: una hora dividida entre el paso te dice cuantas ventanas hay abiertas a la vez por maquina, y cada evento entra en todas. Con paso de 1 minuto son 60; con paso de 5, doce. El estado de las agregaciones (medias, percentiles, correlaciones) va multiplicado por ese numero. El estado del join no, que ese depende del watermark
4. 04.Calcula features: media, p95, max, derivada temporal, ratio frecuencias, correlación
5. 05.Escribe features a tabla Iceberg (gold.machine_features_1h) en cada trigger
6. 06.El modelo de Aitor lee de esa tabla y genera alertas si score > umbral

1# Estructura conceptual del pipeline (Spark Structured Streaming)
2from pyspark.sql import SparkSession
3from pyspark.sql.functions import (
4    window, col, avg, max as spark_max, min as spark_min,
5    percentile_approx, stddev, corr, lag, expr
6)
7from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType
8
9# El pipeline lee de 2 topics y hace join por machine_id dentro de la ventana
10# Luego calcula features y escribe a Iceberg cada 5 minutos
11
12# Input: ~4.411 eventos/segundo (3.600 vibracion + 811 temperatura)
13# Output: ~340 filas por trigger (1 por maquina con features)
14# Estado mantenido: ~25 MB de agregaciones (340 maquinas x 60 ventanas) + ~100 MB
15# de estado del join stream-stream. Unos 125 MB, mas el overhead de Spark.

Antes de escribir nada, una cosa practica: un pipeline de streaming se prueba sin streaming. La logica que de verdad te puedes equivocar -- las ventanas, el watermark, las cuatro features, el join por maquina -- es exactamente la misma leyendo de Kafka que leyendo un fichero. Asi que se empieza al reves de como parece: cargas el parquet de muestra como un DataFrame normal, montas las agregaciones, miras los numeros a ojo y compruebas que el P95 de una maquina sana y el de TMP-0047, que viene con deriva, no se parecen. Cuando eso cuadra, cambias spark.read por spark.readStream, añades el watermark y el trigger, y lo que puede fallar ya solo es lo de streaming: los late data, el tamaño del estado y el checkpoint. Montarlo al reves -- escribir el streaming entero y depurarlo contra un Kafka vivo -- es la manera mas cara que hay de encontrar un parentesis mal puesto.

## Watermarks y late data: por que 2 minutos

En streaming, los eventos no siempre llegan en orden. Un sensor con mala conectividad wifi puede enviar datos con 30 segundos de retraso. El watermark le dice a Spark: "acepta datos que lleguen hasta X minutos tarde, después descártalos". La elección del watermark es un trade-off entre completitud y latencia.

• ▹Watermark 30s: la latencia más baja, pero se cae todo lo que llegue con más de medio minuto de retraso, y eso pasa cada vez que hay un corte de red en planta.
• ▹Watermark 2 min: cuesta dos minutos de latencia y, por lo que se ha visto en esta plataforma, no pierde prácticamente nada. Mira lo que cuenta Kenji aquí abajo.
• ▹Watermark 5 min: en esta planta no aporta nada sobre los 2 minutos y te saca del SLA de Fernando.

Eliges 2 minutos. Kenji confirma: "En 18 meses he visto exactamente 3 eventos llegar con más de 2 minutos de retraso -- los 3 fueron por caidas de red que duraron más de 10 minutos y se reprocessaron después por batch recovery. 2 minutos es el sweet spot."

## Feature 1: Percentil 95 de vibracion (ventana 1h)

El percentil 95 captura los picos anomalos sin ser sensible a outliers extremos (como un golpe accidental). Patricia explica: "La media me oculta los picos. Si una maquina vibra a 2 mm/s el 95% del tiempo pero tiene picos de 15 mm/s cada 5 minutos, la media será 2.5 pero el P95 será 12+. Ese P95 es lo que me dice que algo va mal."

En Spark Structured Streaming, el P95 exacto requiere mantener todos los valores de la ventana en memoria. Con 720 sensores a 5 Hz y ventana de 1 hora, eso son 720 x 5 x 3600 = 12.96 millones de valores. A 8 bytes cada uno, ~100 MB, y creciendo con cada sensor que añadas. Por eso no se guardan: se usa percentile_approx, que no almacena los valores sino un resumen de unos 100 tramos, unos 800 bytes por maquina y metrica. Con 340 maquinas y las sesenta ventanas que el paso de 1 minuto deja abiertas a la vez son unos 16 MB, y -- esto es lo importante -- esa cifra no cambia aunque multipliques por diez los datos de entrada: el resumen ocupa lo mismo con doce millones de valores que con doce mil. Lo que la hace crecer es tener mas maquinas o mas ventanas abiertas, no mas datos.

A cambio, el percentil no es exacto, y conviene entender en que sentido: no te devuelve el P95 con un error de tanto por ciento en el valor, te devuelve un valor que en la ordenacion esta muy cerca del P95 -- puede ser el P94 o el P96. Para decidir si un rodamiento esta degradado, sobra.

Esto tiene nombre y conviene que lo reconozcas cuando te lo encuentres: es un sketch, una estructura que no guarda los datos sino un resumen de ellos con un error acotado. El truco es siempre el mismo: renuncias a la respuesta exacta a cambio de que la memoria deje de crecer con el volumen. Un sketch de percentiles guarda unos cien tramos; uno de cardinalidad (approx_count_distinct) guarda unos kilobytes y te cuenta cuantos sensores distintos han emitido, den igual mil o diez millones. La pregunta que hay que hacerse siempre es si el error te vale. Para decidir si paras una maquina de dos millones de euros, un percentil con un 1% de error vale de sobra. Para contar piezas que se facturan a un cliente, no vale: ahi se cuenta exacto aunque cueste. Esa decision es tuya y es la que hay que dejar escrita.

## Feature 2: Derivada termica (cambio por minuto)

La derivada de temperatura indica la VELOCIDAD de cambio, no el valor absoluto. Un motor a 85C es normal si lleva 4 horas así. Un motor que pasa de 60C a 85C en 15 minutos tiene un problema -- eso es +1.67 C/minuto, cuando el patron normal es <0.2 C/minuto.

Para calcular la derivada, necesitas la temperatura hace 1 minuto. En streaming, eso requiere estado: guardas el valor anterior y calculas (temp_actual - temp_hace_1_min) / 1.0. Usaras una ventana tumbling de 1 minuto para promediar la temperatura del minuto actual y comparar con el promedio del minuto anterior (usando lag sobre el estado).

## Feature 3: Ratio alta/baja frecuencia

El sensor MVM-500 reporta la frecuencia dominante (dominant_freq_hz) -- la calcula el a bordo, analizando la señal a varios kilohercios, y te manda ya el resultado. El husillo de una fresadora gira en torno a los 180 Hz, y sus armonicos salen en los multiplos: 360, 540 y 720. Los tres primeros estan presentes en una maquina perfectamente sana; el que delata el rodamiento es el cuarto, los 720 Hz de Patricia. Asi que el corte va entre el tercero y el cuarto, en 630 Hz: por debajo, vibracion normal de la maquina; por encima, el armonico que nos interesa. El ratio es count(dominant_freq_hz >= 630) / count(eventos_total) en la ventana de 1h.

Fijate en de donde ha salido ese 630: no es un numero redondo elegido a ojo, es 3,5 veces la velocidad de giro, puesto a proposito en el hueco entre dos armonicos. Si mañana instrumentas una prensa que gira a otra velocidad, su corte es otro, y sale de la misma cuenta. Un umbral que no puedes justificar con el proceso que hay detras es un umbral que alguien te va a cambiar en seis meses sin saber por que estaba ahi.

Patricia: "Cuando ese ratio supera 0.3 -- es decir, más del 30% de las lecturas tienen frecuencia dominante alta -- el rodamiento está degradado. Tenemos entre 24 y 48 horas. Si supera 0.5, son menos de 12 horas." Ojo: esos 0,3 y 0,5 son la referencia de partida. Como acabas de cambiar que entra en el numerador (el corte en 630 Hz), el valor del ratio cambia entero, asi que hay que revalidar los dos umbrales contra los fallos historicos de mantenimiento.csv antes de ponerlos en produccion. Recalibrar un umbral cuando cambias su definicion es parte del trabajo.

## Feature 4: Correlacion temperatura-vibracion

La correlación de Pearson entre temperatura y vibración por máquina en la última hora. Si ambas suben juntas (correlación > 0.7), es fricción por desgaste mecánico. Si solo sube vibración pero no temperatura (correlación < 0.3), es desbalanceo del rotor -- menos grave, se puede programar mantenimiento para el próximo fin de semana.

Para calcular la correlación en streaming necesitas el join entre vibración y temperatura por machine_id y ventana temporal. Spark soporta stream-stream joins con watermark -- ambos streams deben tener watermark definido y la condición de join debe incluir una restricción temporal (e.g., abs(t_vib - t_temp) < 5 seconds).