Window functions in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

Output of rank() window function

What is the output of the rank() column after running this Spark code?

Apache Spark

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

spark = SparkSession.builder.getOrCreate()
data = [("A", 100), ("B", 200), ("A", 200), ("B", 100), ("A", 100)]
df = spark.createDataFrame(data, ["category", "value"])
windowSpec = Window.partitionBy("category").orderBy("value")
df = df.withColumn("rank", rank().over(windowSpec))
df.orderBy("category", "value").select("category", "value", "rank").collect()

A [Row(category='A', value=100, rank=1), Row(category='A', value=100, rank=2), Row(category='A', value=200, rank=3), Row(category='B', value=100, rank=1), Row(category='B', value=200, rank=2)]

B [Row(category='A', value=100, rank=1), Row(category='A', value=100, rank=1), Row(category='A', value=200, rank=2), Row(category='B', value=100, rank=1), Row(category='B', value=200, rank=2)]

C [Row(category='A', value=100, rank=1), Row(category='A', value=100, rank=1), Row(category='A', value=200, rank=3), Row(category='B', value=100, rank=1), Row(category='B', value=200, rank=2)]

D [Row(category='A', value=100, rank=1), Row(category='A', value=100, rank=2), Row(category='A', value=200, rank=2), Row(category='B', value=100, rank=1), Row(category='B', value=200, rank=3)]

💡 Hint

Remember that rank() assigns the same rank to ties and skips ranks after ties.

The rank() function assigns the same rank to rows with the same value within each partition. For category 'A', the two rows with value 100 get rank 1, and the next value 200 gets rank 3 (rank 2 is skipped). For category 'B', values 100 and 200 get ranks 1 and 2 respectively.

❓ data_output

intermediate

Count rows per group using window function

What is the output of the count column after running this Spark code?

Apache Spark

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

spark = SparkSession.builder.getOrCreate()
data = [("X", 1), ("X", 2), ("Y", 3), ("Y", 4), ("Y", 5)]
df = spark.createDataFrame(data, ["group", "value"])
windowSpec = Window.partitionBy("group")
df = df.withColumn("count", count("value").over(windowSpec))
df.orderBy("group", "value").select("group", "value", "count").collect()

A [Row(group='X', value=1, count=2), Row(group='X', value=2, count=1), Row(group='Y', value=3, count=3), Row(group='Y', value=4, count=2), Row(group='Y', value=5, count=1)]

B [Row(group='X', value=1, count=1), Row(group='X', value=2, count=2), Row(group='Y', value=3, count=1), Row(group='Y', value=4, count=2), Row(group='Y', value=5, count=3)]

C [Row(group='X', value=1, count=5), Row(group='X', value=2, count=5), Row(group='Y', value=3, count=5), Row(group='Y', value=4, count=5), Row(group='Y', value=5, count=5)]

D [Row(group='X', value=1, count=2), Row(group='X', value=2, count=2), Row(group='Y', value=3, count=3), Row(group='Y', value=4, count=3), Row(group='Y', value=5, count=3)]

💡 Hint

Count over a partition counts all rows in that group for each row.

The count() window function counts the number of rows in each partition. For group 'X' there are 2 rows, so each row shows count 2. For group 'Y' there are 3 rows, so each row shows count 3.

❓ visualization

advanced

Visualize cumulative sum with window function

Which option shows the correct cumulative sum of value within each category when using this Spark code?

Apache Spark

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

spark = SparkSession.builder.getOrCreate()
data = [("A", 10), ("A", 20), ("B", 5), ("B", 15), ("A", 30)]
df = spark.createDataFrame(data, ["category", "value"])
windowSpec = Window.partitionBy("category").orderBy("value").rowsBetween(Window.unboundedPreceding, Window.currentRow)
df = df.withColumn("cumulative_sum", sum("value").over(windowSpec))
df.orderBy("category", "value").select("category", "value", "cumulative_sum").collect()

A [Row(category='A', value=10, cumulative_sum=10), Row(category='A', value=20, cumulative_sum=20), Row(category='A', value=30, cumulative_sum=30), Row(category='B', value=5, cumulative_sum=5), Row(category='B', value=15, cumulative_sum=15)]

B [Row(category='A', value=10, cumulative_sum=10), Row(category='A', value=20, cumulative_sum=30), Row(category='A', value=30, cumulative_sum=60), Row(category='B', value=5, cumulative_sum=5), Row(category='B', value=15, cumulative_sum=20)]

C [Row(category='A', value=10, cumulative_sum=10), Row(category='A', value=20, cumulative_sum=30), Row(category='A', value=30, cumulative_sum=50), Row(category='B', value=5, cumulative_sum=5), Row(category='B', value=15, cumulative_sum=20)]

D [Row(category='A', value=10, cumulative_sum=10), Row(category='A', value=20, cumulative_sum=30), Row(category='A', value=30, cumulative_sum=60), Row(category='B', value=5, cumulative_sum=5), Row(category='B', value=15, cumulative_sum=15)]

💡 Hint

Cumulative sum adds all previous values including the current one in order.

The cumulative sum adds values in order within each category. For 'A', ordered values are 10, 20, 30, sums are 10, 30, 60. For 'B', ordered values 5, 15, sums 5, 20.

🔧 Debug

advanced

Identify error in window function usage

What error does this Spark code raise?

Apache Spark

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

spark = SparkSession.builder.getOrCreate()
data = [(1, 100), (2, 200), (3, 300)]
df = spark.createDataFrame(data, ["id", "value"])
windowSpec = Window.orderBy("id")
df = df.withColumn("prev_value", lag("value", 2).over(windowSpec))
df.show()

A No error, output shows null for first two rows in prev_value

B AnalysisException: Window function lag requires an OVER clause with partitioning

C TypeError: lag() missing required positional argument 'offset'

D RuntimeError: Window specification must include partitionBy clause

💡 Hint

lag() can be used with only orderBy in window specification.

The code runs without error. lag() with offset 2 returns null for first two rows because no previous rows exist at that offset. partitionBy is optional.

🚀 Application

expert

Calculate moving average with sliding window

Given this Spark code, what is the output of the moving_avg column?

Apache Spark

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

spark = SparkSession.builder.getOrCreate()
data = [("2023-01-01", 10), ("2023-01-02", 20), ("2023-01-03", 30), ("2023-01-04", 40), ("2023-01-05", 50)]
df = spark.createDataFrame(data, ["date", "value"])
windowSpec = Window.orderBy("date").rowsBetween(-2, 0)
df = df.withColumn("moving_avg", avg("value").over(windowSpec))
df.orderBy("date").select("date", "value", "moving_avg").collect()

A [Row(date='2023-01-01', value=10, moving_avg=10.0), Row(date='2023-01-02', value=20, moving_avg=15.0), Row(date='2023-01-03', value=30, moving_avg=20.0), Row(date='2023-01-04', value=40, moving_avg=30.0), Row(date='2023-01-05', value=50, moving_avg=40.0)]

B [Row(date='2023-01-01', value=10, moving_avg=10.0), Row(date='2023-01-02', value=20, moving_avg=15.0), Row(date='2023-01-03', value=30, moving_avg=20.0), Row(date='2023-01-04', value=40, moving_avg=30.0), Row(date='2023-01-05', value=50, moving_avg=45.0)]

C [Row(date='2023-01-01', value=10, moving_avg=10.0), Row(date='2023-01-02', value=20, moving_avg=15.0), Row(date='2023-01-03', value=30, moving_avg=20.0), Row(date='2023-01-04', value=40, moving_avg=35.0), Row(date='2023-01-05', value=50, moving_avg=40.0)]

D [Row(date='2023-01-01', value=10, moving_avg=10.0), Row(date='2023-01-02', value=20, moving_avg=20.0), Row(date='2023-01-03', value=30, moving_avg=30.0), Row(date='2023-01-04', value=40, moving_avg=40.0), Row(date='2023-01-05', value=50, moving_avg=50.0)]

💡 Hint

Moving average with rowsBetween(-2, 0) averages current row and two previous rows.

For each row, the moving average is the average of the current value and the two previous values (if they exist). For example, on 2023-01-03, average of values 10, 20, 30 is 20.0.

Practice

1. What is the main purpose of window functions in Apache Spark?

easy

A. To perform calculations across rows related to the current row without collapsing the data

B. To group data and reduce the number of rows

C. To filter rows based on a condition

D. To join two datasets together

Solution

1. Step 1: Understand window function behavior

   Window functions calculate values across related rows but keep all original rows intact.

2. Step 2: Compare with grouping

   Grouping reduces rows by aggregation, but window functions add columns without reducing rows.

3. Final Answer:

   To perform calculations across rows related to the current row without collapsing the data -> Option A

4. Quick Check:

   Window functions add calculations without grouping ≠ A [OK]

Hint: Window functions add columns, grouping reduces rows [OK]

Common Mistakes:

• Confusing window functions with groupBy aggregations
• Thinking window functions filter rows
• Assuming window functions join datasets

2. Which of the following is the correct way to define a window specification in PySpark to partition by 'department' and order by 'salary'?

easy

A. Window.groupBy('department').sortBy('salary')

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

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

D. Window.partition('department').order('salary')

Solution

1. Step 1: Recall PySpark window spec syntax

   The correct methods are partitionBy() and orderBy() chained on Window.

2. Step 2: Check method order and names

   partitionBy comes first, then orderBy; groupBy and sortBy are incorrect here.

3. Final Answer:

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

4. Quick Check:

   Use partitionBy then orderBy methods [OK]

Hint: Use Window.partitionBy().orderBy() exactly [OK]

Common Mistakes:

• Using groupBy instead of partitionBy
• Using sortBy instead of orderBy
• Using incorrect method names like partition or order

3. Given the following PySpark code, what will be the output of the 'rank' column?

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

spark = SparkSession.builder.getOrCreate()
data = [("Alice", "Sales", 5000), ("Bob", "Sales", 4000), ("Charlie", "Sales", 5000)]
df = spark.createDataFrame(data, ["name", "department", "salary"])
windowSpec = Window.partitionBy("department").orderBy("salary")
df = df.withColumn("rank", rank().over(windowSpec))
df.orderBy("salary").show()

medium

A. [ ("Bob", 4000, 1), ("Alice", 5000, 2), ("Charlie", 5000, 2) ]

B. [ ("Bob", 4000, 1), ("Alice", 5000, 3), ("Charlie", 5000, 3) ]

C. [ ("Bob", 4000, 1), ("Alice", 5000, 2), ("Charlie", 5000, 3) ]

D. [ ("Bob", 4000, 1), ("Alice", 5000, 1), ("Charlie", 5000, 1) ]

Solution

1. Step 1: Understand rank() behavior

   Rank assigns the same rank to ties and skips ranks after ties.

2. Step 2: Apply rank over salary ascending

   Bob has salary 4000 ranked 1; Alice and Charlie both have 5000, so both get rank 2.

3. Final Answer:

   [ ("Bob", 4000, 1), ("Alice", 5000, 2), ("Charlie", 5000, 2) ] -> Option A

4. Quick Check:

   rank() ties share rank, next rank skips [OK]

Hint: rank() gives same rank to ties, skips next ranks [OK]

Common Mistakes:

• Confusing rank() with dense_rank()
• Assigning different ranks to tied values
• Not ordering data before applying rank

4. Identify the error in this PySpark code snippet using window functions:

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

windowSpec = Window.partitionBy('department')
df = df.withColumn('row_num', row_number().over(windowSpec.orderBy('salary')))

medium

A. Missing import for Window functions causes error

B. windowSpec.orderBy('salary') is valid; no error in code

C. row_number() cannot be used with partitionBy

D. windowSpec.orderBy('salary') is invalid; orderBy should be called on Window, not on windowSpec

Solution

1. Step 1: Check windowSpec usage

   windowSpec is a WindowSpec object; calling orderBy on it is valid and returns a new WindowSpec.

2. Step 2: Confirm chaining

   Chaining orderBy on windowSpec is allowed to refine the window specification.

3. Final Answer:

   windowSpec.orderBy('salary') is valid; no error in code -> Option B

4. Quick Check:

   WindowSpec supports chaining orderBy to refine window [OK]

Hint: WindowSpec supports chaining orderBy to refine window [OK]

Common Mistakes:

• Calling orderBy on windowSpec instead of Window
• Assuming row_number can't be used with partitionBy
• Forgetting to import Window or functions

5. You have a sales dataset with columns: 'store', 'date', and 'sales'. You want to add a column showing the 3-day moving average of sales per store ordered by date. Which PySpark window function setup and code will achieve this?

hard

A. Use Window.partitionBy('store').orderBy('date').rowsBetween(0, 2) and avg('sales').over(windowSpec)

B. Use Window.partitionBy('store').orderBy('date').rangeBetween(-2, 0) and avg('sales').over(windowSpec)

C. Use Window.orderBy('date').rowsBetween(-3, 0) and avg('sales').over(windowSpec)

D. Use Window.partitionBy('store').orderBy('date').rowsBetween(-2, 0) and avg('sales').over(windowSpec)

Solution

1. Step 1: Define window for 3-day moving average per store

   Partition by 'store' to separate stores, order by 'date' for time sequence, use rowsBetween(-2, 0) for current and previous 2 rows.

2. Step 2: Apply avg() over this window

   Use avg('sales').over(windowSpec) to compute moving average over the defined window.

3. Final Answer:

   Use Window.partitionBy('store').orderBy('date').rowsBetween(-2, 0) and avg('sales').over(windowSpec) -> Option D

4. Quick Check:

   rowsBetween(-2,0) for 3-day window, partition by store [OK]

Hint: Use rowsBetween(-2,0) for 3-row moving average [OK]

Common Mistakes:

• Using rangeBetween instead of rowsBetween for row count
• Not partitioning by store
• Using wrong rowsBetween range (like 0 to 2)