04.6 PySpark Window Functions

Architect goal: Understand window functions as calculations over an ordered subset of rows that preserve row-level detail. Learn partitioning, ordering, frames, ranking, offsets, running calculations, latest-record patterns, and the performance implications of window execution.

1. What is a window function?

An aggregate with groupBy() collapses rows:

100 customer orders
       ↓
1 customer aggregate row

A window calculation normally preserves those 100 rows while adding context calculated from related rows:

100 customer orders
       ↓
100 rows + rank / previous amount / running total / etc.

This distinction is fundamental.

2. Window specification

A window is defined by three concepts:

WindowSpec
 ├── PARTITION BY → which rows belong together
 ├── ORDER BY     → how rows are sequenced
 └── FRAME        → which rows participate for the current row

PySpark exposes partitionBy(), orderBy(), rowsBetween(), and rangeBetween(). Current Spark documentation also defines different default frames depending on whether ordering is present. citeturn440955search0turn440955search1

3. Basic example

from pyspark.sql import Window
from pyspark.sql import functions as F

w = (
    Window
    .partitionBy("customer_id")
    .orderBy(F.col("order_date"), F.col("order_id"))
)

result = (
    orders
    .withColumn("rn", F.row_number().over(w))
)

Interpretation:

For each customer, order the records by date and order_id, then assign a sequential number.

4. Partitioning is not the same as a DataFrame partition

This is a common source of confusion.

Window partition

Window.partitionBy("customer_id")

defines the logical groups used by the window calculation.

Spark data partition

A Spark partition is a physical unit of distributed execution.

A window partition may require Spark to redistribute data so that rows belonging to the same logical group can be processed together.

Keep the concepts separate.

5. Ranking functions

Current PySpark exposes row_number, rank, dense_rank, ntile, percent_rank, and cume_dist, among other window functions. citeturn440955search4turn440955search8

row_number()

Always assigns a sequential number within the ordered partition.

w = Window.partitionBy("department").orderBy(F.col("salary").desc())

df.withColumn(
    "row_number",
    F.row_number().over(w)
)

If two employees have the same salary, they still receive different row numbers. Therefore use a deterministic tie-breaker when selecting a single record.

rank()

Tied values receive the same rank and subsequent ranks can contain gaps.

Salary: 100, 100, 90

rank:     1,   1,  3

dense_rank()

Tied values receive the same rank without gaps.

Salary: 100, 100, 90

dense_rank: 1, 1, 2

6. Deterministic latest-record pattern

This is one of the most important enterprise patterns.

w = (
    Window
    .partitionBy("business_key")
    .orderBy(
        F.col("event_timestamp").desc(),
        F.col("ingestion_sequence").desc()
    )
)

latest = (
    df
    .withColumn("_rn", F.row_number().over(w))
    .filter(F.col("_rn") == 1)
    .drop("_rn")
)

Do not order only by timestamp if timestamps can tie. Add a deterministic secondary key.

This pattern is frequently used for CDC consolidation, latest-state selection, and deduplication.

7. lag() and lead()

lag() looks backward in the ordered partition; lead() looks forward. Current PySpark supports an offset and optional default value. citeturn440955search3turn440955search4

w = Window.partitionBy("customer_id").orderBy("order_date")

result = (
    orders
    .withColumn("previous_amount", F.lag("amount").over(w))
    .withColumn("next_amount", F.lead("amount").over(w))
)

Useful for:

• Period-over-period comparison
• Previous/next event
• State transitions
• Flight sequence analysis
• CDC change detection

8. Calculate change from the previous row

result = (
    orders
    .withColumn("previous_amount", F.lag("amount").over(w))
    .withColumn(
        "amount_change",
        F.col("amount") - F.col("previous_amount")
    )
)

For percentage change, protect against zero denominators.

9. Running totals

w = (
    Window
    .partitionBy("customer_id")
    .orderBy("order_date")
    .rowsBetween(Window.unboundedPreceding, Window.currentRow)
)

result = orders.withColumn(
    "running_revenue",
    F.sum("amount").over(w)
)

This preserves every order while adding cumulative revenue.

10. Window frames

Frames control which rows are visible to the current calculation.

ROWS frame

A row-based frame uses physical row positions within the ordered partition.

w.rowsBetween(-2, Window.currentRow)

means:

current row + the previous two rows.

Spark documents row boundaries as positions relative to the current row. citeturn440955search10

RANGE frame

A range-based frame is based on the value of the ORDER BY expression rather than row position.

This distinction matters when there are duplicate order values.

Spark documents restrictions on bounded range frames, including the ordering expression requirements. citeturn440955search0

11. ROWS vs RANGE

Consider:

date       amount
Jan 01     100
Jan 01     200
Jan 02     300

A ROWS frame counts row positions.

A RANGE frame evaluates the ORDER BY value and may include multiple peer rows with the same ordering value.

Architect rule

Never choose ROWS or RANGE because the syntax “looks right.” Define the business meaning of the frame first.

12. Moving average

w = (
    Window
    .partitionBy("customer_id")
    .orderBy("order_date")
    .rowsBetween(-6, Window.currentRow)
)

result = orders.withColumn(
    "7_row_avg",
    F.avg("amount").over(w)
)

This is a seven-row moving average, not necessarily a seven-day average.

For a true time-based window, use an appropriate RANGE frame or a date/calendar modeling approach.

13. Previous and next business state

w = (
    Window
    .partitionBy("flight_id")
    .orderBy("event_timestamp")
)

events = (
    events
    .withColumn("previous_status", F.lag("status").over(w))
    .withColumn("next_status", F.lead("status").over(w))
)

This supports event-sequence analysis.

14. First and last values

Windowed first and last can be useful, but frame boundaries matter.

For example, “last value in the entire partition” is different from “last value seen up to the current row.”

Always specify or inspect the intended frame.

15. Top-N per group

A classic enterprise requirement:

Find the top 3 products by revenue for each country.

w = (
    Window
    .partitionBy("country")
    .orderBy(F.col("revenue").desc(), F.col("product_id"))
)

top3 = (
    product_sales
    .withColumn("_rank", F.row_number().over(w))
    .filter(F.col("_rank") <= 3)
    .drop("_rank")
)

Choose row_number, rank, or dense_rank according to how ties should behave.

16. Window vs groupBy

Requirement	groupBy	Window
One row per customer	Yes	Usually unnecessary
Keep every order	No	Yes
Customer total repeated on every order	No	Yes
Latest row per key	Possible but awkward	Excellent
Running total	No	Yes
Previous event	No	Yes
Top N per group	Possible but complex	Excellent

17. Window execution can be expensive

Window calculations often require data to be colocated by partition keys and ordered according to the window specification.

A conceptual physical flow is:

Scan
  ↓
Filter / Project
  ↓
Exchange by window partition keys
  ↓
Sort by window ordering
  ↓
Window evaluation
  ↓
Result

Exact physical operators vary by query and runtime.

18. The biggest performance risk: huge logical partitions

Consider:

Window.partitionBy("customer_id")

If one customer owns 40% of the entire dataset, the window partition for that customer is enormous.

Possible symptoms:

• One or a few long-running tasks
• Large shuffle
• Spill
• Memory pressure
• Stragglers

Window skew is therefore a data-distribution problem.

19. Avoid global windows unless required

This:

Window.orderBy("event_timestamp")

has no partition key.

It can force global ordering semantics and become extremely expensive on large datasets.

Ask whether the calculation really needs one global sequence.

20. Reduce data before the window

Bad:

w = Window.partitionBy("customer_id").orderBy("order_date")
df.withColumn("rn", F.row_number().over(w))

when df contains hundreds of irrelevant columns and years of data.

Better:

input_df = (
    df
    .select("customer_id", "order_id", "order_date", "amount")
    .filter(F.col("order_date") >= F.lit("2026-01-01"))
)

result = input_df.withColumn("rn", F.row_number().over(w))

Reduce rows and columns before expensive window processing.

21. Determinism matters

This is dangerous:

Window.partitionBy("customer_id").orderBy("event_timestamp")

if multiple events share the same timestamp and the business requires a unique winner.

Prefer:

Window.partitionBy("customer_id").orderBy(
    F.col("event_timestamp").desc(),
    F.col("event_id").desc()
)

A deterministic ordering is part of data correctness.

22. Window functions for CDC

Suppose a CDC stream contains multiple changes for a business key.

Use:

w = (
    Window
    .partitionBy("business_key")
    .orderBy(
        F.col("change_timestamp").desc(),
        F.col("sequence_number").desc()
    )
)

current_state = (
    cdc
    .withColumn("_rn", F.row_number().over(w))
    .filter(F.col("_rn") == 1)
)

Then apply the resulting current state to a target table.

This is a reusable Bronze → Silver pattern.

23. Window functions for SCD2 validation

Windows can help identify overlapping or unexpected records:

w = (
    Window
    .partitionBy("customer_id")
    .orderBy("valid_from")
)

check = (
    dim
    .withColumn("next_valid_from", F.lead("valid_from").over(w))
)

Then compare valid_to with the next valid_from.

This is especially useful for temporal data quality.

24. Flight analytics example

For an airline dataset:

w = (
    Window
    .partitionBy("aircraft_id")
    .orderBy("departure_timestamp")
)

flight_sequence = (
    flights
    .withColumn("previous_arrival", F.lag("arrival_timestamp").over(w))
    .withColumn("next_departure", F.lead("departure_timestamp").over(w))
)

Possible metrics:

• Aircraft turnaround time
• Previous-leg delay
• Next-leg delay
• Sequence gaps
• Aircraft utilization

25. Performance troubleshooting

When a window job is slow:

1. Inspect the logical requirement

Does the calculation really need a window?

2. Reduce input

Filter and project early.

3. Inspect partition cardinality

How many rows are in each logical window partition?

4. Inspect physical plan

result.explain("formatted")

Look for exchange and sorting.

5. Inspect Spark UI

Check:

• Shuffle read/write
• Task-duration distribution
• Spill
• Skew
• Failed/retried tasks

6. Check ordering

Can the order be simplified or made deterministic with a narrow key?

7. Reconsider the model

Would a pre-aggregated Gold table remove the need for a large window?

26. Common anti-patterns

Anti-pattern 1 — Global row_number

Using:

Window.orderBy("timestamp")

on billions of rows without needing global ordering.

Anti-pattern 2 — Huge window partitions

Partitioning by a key with extreme skew.

Anti-pattern 3 — Non-deterministic latest-row selection

Ordering only by a timestamp that can tie.

Anti-pattern 4 — Carrying unnecessary columns

Wide rows increase shuffle and sort cost.

Anti-pattern 5 — Window when groupBy is enough

If the result only needs one row per group, use an aggregation rather than preserving every row.

27. Hands-on lab — latest record

Create multiple versions of each customer record.

Implement:

1. Latest record using row_number().
2. Deterministic tie-breaking.
3. Compare with a non-deterministic ordering.
4. Validate one output row per business key.

28. Hands-on lab — running revenue

Create order data and calculate:

• Daily revenue
• Customer running revenue
• Previous order amount
• Change from previous order
• Seven-row moving average

Then inspect the plan.

29. Hands-on lab — top N

Build:

Top 3 products by revenue for each country.

Compare:

• row_number()
• rank()
• dense_rank()

Create ties and observe the different outputs.

30. Hands-on lab — window skew

Generate a dataset where one customer owns most rows.

Run a window calculation and compare task-duration distribution with a balanced dataset.

Identify:

• Straggler tasks
• Spill
• Shuffle imbalance
• Memory pressure

31. Architect decision framework

Requirement	Recommended starting point
One row per group	groupBy
Latest record per key	row_number
Top N with unique positions	row_number
Ranking with gaps for ties	rank
Ranking without gaps	dense_rank
Previous row	lag
Next row	lead
Running metric	ordered window + frame
Moving row-based metric	ROWS frame
Value-based range metric	RANGE frame
Global ordering	Avoid unless required
Large/skewed partition	Reconsider design

32. Interview questions

1. What is a window function?
2. Window vs groupBy?
3. What does partitionBy mean?
4. Is a window partition the same as a Spark partition?
5. row_number vs rank vs dense_rank?
6. Why can latest-record logic be non-deterministic?
7. lag vs lead?
8. ROWS vs RANGE?
9. Why can windows be expensive?
10. What happens if partitionBy is omitted?
11. How do you troubleshoot a slow window?
12. How can window functions support CDC?
13. How can windows help validate SCD2?
14. How would you implement top 3 per country?
15. How would you handle a heavily skewed window partition?

33. Cheat sheet

from pyspark.sql import Window
from pyspark.sql import functions as F

# Partition + order
w = Window.partitionBy("customer_id").orderBy("order_date")

# Row number
F.row_number().over(w)

# Rank
F.rank().over(w)

# Dense rank
F.dense_rank().over(w)

# Previous / next
F.lag("amount").over(w)
F.lead("amount").over(w)

# Running total
running = (
    w.rowsBetween(
        Window.unboundedPreceding,
        Window.currentRow
    )
)

F.sum("amount").over(running)

# Top N
df.withColumn("_rn", F.row_number().over(w)).filter(F.col("_rn") <= 3)

# Inspect
df.explain("formatted")

34. Key takeaways

1. A window adds context without normally collapsing row-level detail.
2. Every window has partitioning, ordering, and potentially a frame.
3. Window partitioning is a logical concept; Spark partitions are physical execution units.
4. row_number, rank, and dense_rank have different tie semantics.
5. lag and lead are fundamental for sequence and change analysis.
6. ROWS and RANGE do not mean the same thing.
7. Deterministic ordering is essential for reliable latest-row logic.
8. Large or skewed window partitions can become expensive.
9. Global windows should be used only when global semantics are actually required.
10. Performance tuning starts by reducing data and understanding the physical plan.

Next

04.7 PySpark UDFs — Python UDFs, pandas UDFs, native Spark expressions, Arrow, serialization boundaries, performance, when UDFs are justified, and production alternatives.

Official references

• PySpark Window
• PySpark Window rowsBetween
• PySpark functions — window functions
• Databricks PySpark functions