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 80 
20     en 10 minutos tiene un problema.
21
223. Ratio alta/baja frecuencia de vibracion
23   - Cuando el armonico de fallo (4x husillo) supera el 30% del 
24     fundamental, hay degradación. Necesitamos ese ratio.
25
264. Correlacion cruzada temperatura-vibracion por maquina (1h)
27   - Si temperatura y vibracion suben juntas = friccion por desgaste.
28   - Si solo sube vibracion pero no temp = desbalanceo (menos grave).
29
30Con estas 4 features, en nuestras pruebas offline subimos de 72% a 94%."
31
32Fernando (que se ha asomado): "Necesito ese pipeline de features en 
33producción en 2 semanas. Que las features se calculen en streaming 
34con ventana de 1 hora y esten disponibles para el modelo en menos de 
355 minutos desde que el dato entra. Es la prioridad número 1."

### 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 680 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.

### 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 (sliding, step de 5 min)
4. 04.Calcula features: media, p95, max, derivada temporal, ratio frecuencias, correlación
5. 05.Escribe features a tabla Iceberg (gold.machine_features_1h) cada 5 minutos
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.280 eventos/segundo (3.600 vibracion + 680 temperatura)
13# Output: ~340 filas cada 5 minutos (1 por maquina con features)
14# Estado mantenido: ~1 hora de datos por maquina en memoria (~800 MB total)

### 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: baja latencia pero pierde ~2% de eventos (los que llegan con más de 30s de retraso por wifi en planta)
• ▹Watermark 2 min: pierde <0.01% de eventos (solo fallos de red graves). Anade 2 min de latencia al resultado.
• ▹Watermark 5 min: no pierde nada en condiciones normales pero la latencia sube a 7 min (fuera 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. Manejable, pero usaremos percentile_approx con precisión de 100 buckets para reducir a ~30 MB de estado sin perder precisión significativa (error < 1%).

### 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). Si es >1000 Hz, es "alta frecuencia" (armonicos de fallo). Si es <500 Hz, es "baja frecuencia" (vibracion fundamental normal). El ratio es: count(eventos_alta_freq) / count(eventos_total) en la ventana de 1h.

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."

### 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).