How to calculate rolling averages in Apache Spark: step by step

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

This tutorial shows how to calculate rolling averages (moving averages) in Apache Spark for time series, useful to smooth fluctuations and detect trends. I explain the rationale for the Window approach and present a practical example in PySpark, including performance handling and common pitfalls.

Prerequisites

Installation of Apache Spark (or an environment like Databricks) and Python with PySpark.
Basic knowledge of DataFrame in PySpark (select, withColumn, groupBy).
Time series dataset with datetime and value columns.

Step 1: Understand the goal and choose the window

A rolling average computes the average of the last N points for each record. It is important to decide whether the window is by number of rows (e.g., 7 days) or by time interval (e.g., last 7 days) — this affects how we define the Window and the ordering.

Step 2: Read the data and prepare timestamps

Read the data into a DataFrame and ensure the date column is of Timestamp/Date type. Ordering by timestamp is crucial for a correct moving average.

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

Step 3: Moving average by fixed number of rows (e.g., average of last 7 records)

For a window based on number of records we use Window.rowsBetween with negative offsets. Useful when the data has regular intervals or when we want N records before the current one.

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 points: current + 6 previous

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

Step 4: Moving average by time interval (e.g., last 7 days)

When records are irregular, we prefer a time-based window with Window.rangeBetween in milliseconds. First obtain the timestamp as 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)

Step 5: Performance and optimizations

Windows can be expensive. Tips: partition by a relevant key (e.g., sensor_id) for parallelism, reduce columns with select, and use caching when reapplying windows. Avoid global orderBy if not necessary.

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

Step 6: Common mistakes and how to avoid them

Frequent mistakes: using rowsBetween for time-based windows (leads to wrong results with irregular records), forgetting to order, and not partitioning when there are many keys. Check column types and convert to timestamp/long before using rangeBetween.

Verify the result

Confirm integrity by comparing some points with manual calculation or using a reduced window. Check for null values in the first N rows (when there are not enough previous points) and confirm that the average for a point matches the average of the records within the time or row window.

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

Conclusion

Rolling averages in Apache Spark help detect trends in time series; choose between row-based or time-based windows depending on the nature of the data. Next steps: experiment with exponential windows, combine with aggregate functions (stddev) or use state with Structured Streaming. Tip: start with small samples and validate with manual calculation before applying to the entire dataset.