Como calcular rolling averages em Apache Spark: passo a passo

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

Este tutorial mostra como calcular rolling averages (médias móveis) em Apache Spark para séries temporais, útil para suavizar flutuações e detetar tendências. Explico a razão da abordagem com Window e apresento um exemplo prático em PySpark, incluindo tratamento de performance e erros comuns.

Pré-requisitos

Instalação do Apache Spark (ou ambiente como Databricks) e Python com PySpark.
Conhecimentos básicos de DataFrame em PySpark (select, withColumn, groupBy).
Conjunto de dados de série temporal com colunas de data/tempo e valor.

Passo 1: Entender o objetivo e escolher a janela

Uma rolling average calcula a média dos últimos N pontos para cada registo. É importante decidir se a janela é por número de linhas (ex.: 7 dias) ou por intervalo temporal (ex.: últimos 7 dias) — isso afecta como definimos a Window e a ordenação.

Passo 2: Ler os dados e preparar timestamps

Ler os dados para um DataFrame e garantir que a coluna de data está em tipo Timestamp/Date. Ordenar por timestamp é crucial para uma média móvel 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()

Passo 3: Média móvel por número fixo de linhas (ex.: média dos últimos 7 registos)

Para uma janela baseada em número de registos usamos Window.rowsBetween com offsets negativos. Útil quando os dados têm intervalos regulares ou quando queremos N registos antes do actual.

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

# Ordenar por event_time e, se houver, por id para determinismo
w = Window.partitionBy().orderBy(col("event_time")).rowsBetween(-6, 0)  # 7 pontos: 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)

Passo 4: Média móvel por intervalo temporal (ex.: últimos 7 dias)

Quando os registos são irregulares, preferimos uma janela por intervalo de tempo com Window.rangeBetween em milissegundos. Primeiro obtemos o 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 dias = 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)

Passo 5: Performance e optimizações

Windows podem ser dispendiosos. Dicas: particionar por chave relevante (ex.: sensor_id) para paralelismo, reduzir colunas com select, e usar caching quando reaplicar janelas. Evitar orderBy global se não for necessário.

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

Passo 6: Erros comuns e como evitá-los

Erros frequentes: usar rowsBetween para janelas temporais (conduz a resultados errados com registos irregulares), esquecer de ordenar, e não particionar quando há muitas chaves. Verificar tipos de coluna e converter para timestamp/long antes de usar rangeBetween.

Verificar o resultado

Confirmar integridade comparando alguns pontos com cálculo manual ou usando uma janela reduzida. Verifique valores nulos nas primeiras N linhas (quando não há pontos anteriores suficientes) e confirme que a média para um ponto corresponde à média dos registos dentro da janela temporal ou de linhas.

# 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"]])

Conclusão

As rolling averages em Apache Spark ajudam a detetar tendências em séries temporais; escolha entre janelas por linhas ou por tempo conforme a natureza dos dados. Próximos passos: experimentar janelas exponenciais, combinar com funções agregadas (stddev) ou usar estados com Structured Streaming. Dica: comece por pequenas amostras e valide com cálculo manual antes de aplicar a toda a base.