Windowed aggregations in Apache Spark

Introduction

Windowed aggregations help us calculate values like sums or averages over a moving group of rows. This lets us see trends or comparisons within parts of our data.

To find a running total of sales for each day in a month.

To compare each student's score with the average score in their class.

To calculate the average temperature over the last 7 days for each city.

To rank products by sales within each category.

To find the difference between each row's value and the previous row's value.

Syntax

Apache Spark

from pyspark.sql import Window
from pyspark.sql.functions import sum, avg, rank

windowSpec = Window.partitionBy('group_column').orderBy('order_column').rowsBetween(start, end)

# Example aggregation
df.withColumn('running_sum', sum('value_column').over(windowSpec))

partitionBy groups data into windows.

orderBy defines the order inside each group.

Examples

This calculates the total sales up to the current date for each category.

Apache Spark

from pyspark.sql import Window
from pyspark.sql.functions import sum

windowSpec = Window.partitionBy('category').orderBy('date').rowsBetween(Window.unboundedPreceding, 0)
df.withColumn('cumulative_sales', sum('sales').over(windowSpec))

This finds the average temperature over the last 7 days for each city.

Apache Spark

from pyspark.sql import Window
from pyspark.sql.functions import avg

windowSpec = Window.partitionBy('city').orderBy('date').rowsBetween(-6, 0)
df.withColumn('weekly_avg_temp', avg('temperature').over(windowSpec))

This ranks employees by salary within each department.

Apache Spark

from pyspark.sql import Window
from pyspark.sql.functions import rank, desc

windowSpec = Window.partitionBy('department').orderBy(desc('salary'))
df.withColumn('salary_rank', rank().over(windowSpec))

Sample Program

This program creates a small sales dataset. It then calculates the running total of sales for each category by date using windowed aggregation.

Apache Spark

from pyspark.sql import SparkSession
from pyspark.sql import Window
from pyspark.sql.functions import sum, avg

spark = SparkSession.builder.appName('WindowedAggregationsExample').getOrCreate()

# Sample data
data = [
    ('A', '2024-01-01', 100),
    ('A', '2024-01-02', 150),
    ('A', '2024-01-03', 200),
    ('B', '2024-01-01', 80),
    ('B', '2024-01-02', 120),
    ('B', '2024-01-03', 160)
]

columns = ['category', 'date', 'sales']
df = spark.createDataFrame(data, columns)

# Define window: group by category, order by date, from start to current row
windowSpec = Window.partitionBy('category').orderBy('date').rowsBetween(Window.unboundedPreceding, 0)

# Calculate cumulative sales
result_df = df.withColumn('cumulative_sales', sum('sales').over(windowSpec))

result_df.show()

Important Notes

Window functions do not reduce the number of rows; they add new columns with aggregated values.

Ordering inside the window is important for cumulative or running calculations.

Use rowsBetween to control the range of rows included in the window.

Summary

Windowed aggregations let you calculate values over groups of rows without losing detail.

They are useful for running totals, moving averages, rankings, and comparisons.

Use partitionBy to group, orderBy to sort, and rowsBetween to set the window range.

Practice

1. What is the main purpose of using windowed aggregations in Apache Spark?

easy

A. To perform calculations over a group of rows without collapsing them into a single result

B. To join two DataFrames based on a key column

C. To filter rows based on a condition

D. To sort the entire DataFrame globally

Solution

1. Step 1: Understand windowed aggregation concept

   Windowed aggregations allow calculations like sums or averages over a set of rows related to the current row, without reducing the number of rows.

2. Step 2: Compare with other operations

   Joins combine tables, filters remove rows, and sorting orders rows globally, but do not perform grouped calculations without collapsing rows.

3. Final Answer:

   To perform calculations over a group of rows without collapsing them into a single result -> Option A

4. Quick Check:

   Windowed aggregation = grouped calculation without row collapse [OK]

Hint: Windowed means grouped but keeps all rows [OK]

Common Mistakes:

• Confusing windowed aggregation with join operations
• Thinking windowed aggregation filters rows
• Assuming windowed aggregation sorts globally

2. Which of the following is the correct way to define a window specification that partitions data by 'department' and orders by 'salary' in ascending order?

easy

A. Window.orderBy('department').partitionBy('salary')

B. Window.partitionBy('department').orderBy('salary')

C. Window.partitionBy('salary').orderBy('department')

D. Window.orderBy('salary').partitionBy('department')

Solution

1. Step 1: Recall window specification syntax

   In Spark, window specs use partitionBy first to group, then orderBy to sort within groups.

2. Step 2: Check option order and columns

   Window.partitionBy('department').orderBy('salary') correctly partitions by 'department' and orders by 'salary'. Other options reverse or swap columns incorrectly.

3. Final Answer:

   Window.partitionBy('department').orderBy('salary') -> Option B

4. Quick Check:

   partitionBy then orderBy = correct window spec [OK]

Hint: partitionBy groups first, orderBy sorts inside groups [OK]

Common Mistakes:

• Swapping partitionBy and orderBy order
• Using wrong columns in partitionBy or orderBy
• Confusing partitionBy with filter

3. Given the following Spark code, what will be the output for the 'running_total' column?

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import sum

spark = SparkSession.builder.getOrCreate()
data = [(1, 'A', 10), (2, 'A', 20), (3, 'A', 30), (4, 'B', 40)]
df = spark.createDataFrame(data, ['id', 'group', 'value'])

windowSpec = Window.partitionBy('group').orderBy('id').rowsBetween(Window.unboundedPreceding, Window.currentRow)
df = df.withColumn('running_total', sum('value').over(windowSpec))
df.orderBy('id').show()

medium

A. [10, 30, 60, 100]

B. [10, 20, 30, 40]

C. [10, 30, 60, 40]

D. [10, 20, 50, 40]

Solution

1. Step 1: Understand window specification

   The window partitions by 'group' and orders by 'id'. The frame includes all rows from the start to the current row in that group.

2. Step 2: Calculate running totals per group

   For group 'A': id=1 value=10 running_total=10; id=2 value=20 running_total=10+20=30; id=3 value=30 running_total=10+20+30=60. For group 'B': id=4 value=40 running_total=40.

3. Final Answer:

   [10, 30, 60, 40] -> Option C

4. Quick Check:

   Running total sums values up to current row per group [OK]

Hint: Running total sums values up to current row in partition [OK]

Common Mistakes:

• Not partitioning by group, summing all rows
• Misunderstanding rowsBetween frame
• Mixing order of rows in calculation

4. Identify the error in the following Spark code snippet that tries to calculate a moving average over the last 2 rows including the current row:

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

windowSpec = Window.partitionBy('category').orderBy('date').rowsBetween(-2, 0)
df = df.withColumn('moving_avg', avg('sales').over(windowSpec))

medium

A. partitionBy should be orderBy for moving average

B. orderBy column 'date' must be included in select

C. avg function cannot be used with window functions

D. rowsBetween frame is incorrect; it should be rowsBetween(-1, 0) for last 2 rows

Solution

1. Step 1: Understand rowsBetween frame

   rowsBetween(-2, 0) includes current row and two preceding rows, totaling 3 rows, not 2.

2. Step 2: Correct frame for last 2 rows

   To include current and previous row (2 rows total), use rowsBetween(-1, 0).

3. Final Answer:

   rowsBetween frame is incorrect; it should be rowsBetween(-1, 0) for last 2 rows -> Option D

4. Quick Check:

   rowsBetween(-1,0) = last 2 rows including current [OK]

Hint: rowsBetween counts rows including current, so -1 means one before [OK]

Common Mistakes:

• Using wrong rowsBetween range for desired window size
• Confusing partitionBy and orderBy roles
• Thinking avg can't be used with window

5. You have a sales DataFrame with columns: 'store', 'date', and 'sales'. You want to calculate a 7-day moving average of sales per store, but some dates are missing for certain stores. Which approach correctly handles missing dates to compute the moving average using windowed aggregations?

hard

A. Fill missing dates with zero sales before applying a window partitioned by 'store' and ordered by 'date' with rowsBetween(-6, 0)

B. Use partitionBy('store') and orderBy('date') with rowsBetween(-7, 0) directly on original data without filling missing dates

C. Apply a global window without partitioning and order by 'date' only

D. Calculate moving average using groupBy('store', 'date') and avg('sales') without window functions

Solution

1. Step 1: Understand missing dates impact

   Missing dates cause gaps in ordering, so rowsBetween(-6, 0) won't cover full 7 days if dates are missing.

2. Step 2: Fill missing dates with zero sales

   Filling missing dates ensures continuous date sequence per store, so window frame covers exact 7 days.

3. Step 3: Apply window with partition and order

   Partition by 'store', order by 'date', and use rowsBetween(-6, 0) to get 7-day moving average correctly.

4. Final Answer:

   Fill missing dates with zero sales before applying a window partitioned by 'store' and ordered by 'date' with rowsBetween(-6, 0) -> Option A

5. Quick Check:

   Fill gaps first, then window for accurate moving average [OK]

Hint: Fill missing dates to avoid gaps in window frame [OK]

Common Mistakes:

• Ignoring missing dates causing wrong window frames
• Using global window ignoring store partitions
• Using groupBy instead of window for moving average