Cómo calcular rolling averages en Apache Spark: paso a paso

João Barros 16 de August de 2026 4 min de lectura

Este tutorial muestra cómo calcular rolling averages (medias móviles) en Apache Spark para series temporales, útil para suavizar fluctuaciones y detectar tendencias. Explico la razón del enfoque con Window y presento un ejemplo práctico en PySpark, incluyendo tratamiento de rendimiento y errores comunes.

Prerequisitos

Instalación de Apache Spark (o entorno como Databricks) y Python con PySpark.
Conocimientos básicos de DataFrame en PySpark (select, withColumn, groupBy).
Conjunto de datos de serie temporal con columnas de fecha/hora y valor.

Paso 1: Entender el objetivo y elegir la ventana

Una rolling average calcula la media de los últimos N puntos para cada registro. Es importante decidir si la ventana es por número de filas (ej.: 7 días) o por intervalo temporal (ej.: últimos 7 días) — esto afecta cómo definimos la Window y el ordenamiento.

Paso 2: Leer los datos y preparar timestamps

Leer los datos en un DataFrame y garantizar que la columna de fecha está en tipo Timestamp/Date. Ordenar por timestamp es crucial para una media móvil correcta.

from pyspark.sql import SparkSession
from pyspark.sql.functions import to_timestamp

spark = SparkSession.builder.appName("rolling_avg").getOrCreate()
# Exemplo: ficheiro CSV com colunas: id, event_time, value
df = spark.read.csv("/path/data.csv", header=True, inferSchema=True)
# Converter para timestamp se necessário
df = df.withColumn("event_time", to_timestamp("event_time"))
# Verificar esquema
df.printSchema()

Paso 3: Media móvil por número fijo de filas (ej.: media de los últimos 7 registros)

Para una ventana basada en número de registros usamos Window.rowsBetween con offsets negativos. Útil cuando los datos tienen intervalos regulares o cuando queremos N registros antes del actual.

from pyspark.sql.window import Window
from pyspark.sql.functions import avg, col

# Ordenar por event_time y, si hay, por id para determinismo
w = Window.partitionBy().orderBy(col("event_time")).rowsBetween(-6, 0)  # 7 puntos: actual + 6 anteriores

df_mv = df.withColumn("rolling_avg_7", avg(col("value")).over(w))
df_mv.select("event_time", "value", "rolling_avg_7").show(10)

Paso 4: Media móvil por intervalo temporal (ej.: últimos 7 días)

Cuando los registros son irregulares, preferimos una ventana por intervalo de tiempo con Window.rangeBetween en milisegundos. Primero obtenemos el timestamp como long (epoch millis).

from pyspark.sql.functions import unix_timestamp

# Criar coluna epoch em segundos ou milissegundos
df_ts = df.withColumn("ts", unix_timestamp(col("event_time")) * 1000)
# Janela: últimos 7 días = 7 * 24 * 60 * 60 * 1000 ms
seven_days_ms = 7 * 24 * 60 * 60 * 1000
w_time = Window.partitionBy().orderBy(col("ts")).rangeBetween(-seven_days_ms, 0)

df_mv_time = df_ts.withColumn("rolling_avg_7d", avg(col("value")).over(w_time))
df_mv_time.select("event_time", "value", "rolling_avg_7d").show(10)

Paso 5: Rendimiento y optimizaciones

Las Windows pueden ser costosas. Consejos: particionar por clave relevante (ej.: sensor_id) para paralelismo, reducir columnas con select, y usar caching cuando se reapliquen ventanas. Evitar orderBy global si no es necesario.

# Exemplo: particionar por sensor_id para calcular média por sensor
w_sensor = Window.partitionBy("sensor_id").orderBy(col("ts")).rangeBetween(-seven_days_ms, 0)
df_opt = df_ts.select("sensor_id", "ts", "value").withColumn("rolling_avg_7d", avg("value").over(w_sensor))
# Cache se for reutilizado
df_opt.cache()

Paso 6: Errores comunes y cómo evitarlos

Errores frecuentes: usar rowsBetween para ventanas temporales (conduce a resultados incorrectos con registros irregulares), olvidar ordenar, y no particionar cuando hay muchas claves. Verificar tipos de columna y convertir a timestamp/long antes de usar rangeBetween.

Verificar el resultado

Confirmar integridad comparando algunos puntos con cálculo manual o usando una ventana reducida. Verifique valores nulos en las primeras N filas (cuando no hay puntos anteriores suficientes) y confirme que la media para un punto corresponde a la media de los registros dentro de la ventana temporal o de filas.

# Comparação manual para um registo específico (exemplo em Pandas para validação local)
sample = df_mv_time.orderBy("event_time").limit(20).toPandas()
# Calcular média móvel em Pandas para checagem
sample["check_avg_7d"] = sample["value"].rolling(window=7, min_periods=1).mean()
print(sample[["event_time","value","rolling_avg_7d","check_avg_7d"]])

Conclusión

Las rolling averages en Apache Spark ayudan a detectar tendencias en series temporales; elija entre ventanas por filas o por tiempo según la naturaleza de los datos. Próximos pasos: experimentar con ventanas exponenciales, combinar con funciones agregadas (stddev) o usar estados con Structured Streaming. Consejo: comience con pequeñas muestras y valide con cálculo manual antes de aplicar a toda la base.