DataDank homeDataDank home
Menu
SQLPythonPySparkApache SparkKafkaAirflow
PySpark courseLesson 7 of 10

PySpark course · Lesson 7 of 10

PySpark Window Functions: Ranking, Lag and Running Totals

Use PySpark window functions for top-N per group, deduplication, lag and lead, running and rolling totals, gap filling and sessions, and avoid the single-partition trap.

  • Intermediate
  • 18 min read
  • Updated Oct 2026
On this page
  1. Sample data
  2. Window functions in Spark
  3. What a window is
  4. Ranking: row_number, rank and dense_rank
  5. Top N per group
  6. Deduplicate: keep the latest record per key
  7. Previous and next rows: lag and lead
  8. Running totals and the default frame
  9. Rolling windows: ROWS vs RANGE
  10. first, last and share of total
  11. Forward-fill missing values
  12. Sessions: gaps and islands
  13. The same thing in Spark SQL
  14. How a window executes
  15. Pitfalls
  16. In interviews
  17. Practice questions
  18. Key takeaways

Window functions compute a value for each row from a group of related rows without collapsing the rows, which is the difference from groupBy. They answer the questions Data Engineers meet every week: the latest record per key, the top N per group, the change since the previous event, a running or rolling total, and where one session ends and the next begins. The ideas are identical to SQL window functions; this lesson shows the PySpark API, its defaults and what windows cost on a cluster.

Sample data

Rep-level sales with deliberate ties (three sales of 150 in the north, two of 200 in the south) and gaps between dates.

import re, io, contextlib
from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F

spark = (SparkSession.builder.master("local[2]").appName("windows")
         .config("spark.sql.shuffle.partitions", "4").getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

sales = spark.createDataFrame(
    [("Asha", "north", "2026-01-01", 100), ("Asha", "north", "2026-01-02", 150),
     ("Asha", "north", "2026-01-05", 150), ("Ben",  "north", "2026-01-01", 150),
     ("Ben",  "north", "2026-01-03", 90),  ("Chen", "south", "2026-01-01", 200),
     ("Chen", "south", "2026-01-02", 50),  ("Dara", "south", "2026-01-04", 200)],
    "rep STRING, region STRING, sale_date STRING, amount INT",
).withColumn("sale_date", F.to_date("sale_date"))
sales.show()
+----+------+----------+------+
| rep|region| sale_date|amount|
+----+------+----------+------+
|Asha| north|2026-01-01|   100|
|Asha| north|2026-01-02|   150|
|Asha| north|2026-01-05|   150|
| Ben| north|2026-01-01|   150|
| Ben| north|2026-01-03|    90|
|Chen| south|2026-01-01|   200|
|Chen| south|2026-01-02|    50|
|Dara| south|2026-01-04|   200|
+----+------+----------+------+

Window functions in Spark

What a window is

A window function needs three pieces, described by a Window specification and applied with .over(spec):

Piece Method Meaning
Partition Window.partitionBy(cols) Which rows belong together (like GROUP BY, but rows are kept)
Order .orderBy(cols) The order of rows inside each partition
Frame .rowsBetween(start, end) / .rangeBetween(start, end) Which rows around the current one an aggregate sees

There are three families of functions:

  • Ranking: row_number, rank, dense_rank, percent_rank, ntile, cume_dist. Need an orderBy; ignore any frame.
  • Offset (analytic): lag, lead, nth_value, first, last. Look at other rows by position.
  • Aggregate: sum, avg, min, max, count, collect_list and so on, computed over the frame.

Ranking: row_number, rank and dense_rank

by_amount = Window.partitionBy("region").orderBy(F.desc("amount"))
by_amount_unique = Window.partitionBy("region").orderBy(F.desc("amount"), "rep", "sale_date")
sales.select("region", "rep", "sale_date", "amount",
    F.row_number().over(by_amount_unique).alias("row_number"),
    F.rank().over(by_amount).alias("rank"),
    F.dense_rank().over(by_amount).alias("dense_rank"),
    F.round(F.percent_rank().over(by_amount), 2).alias("pct_rank"),
    F.ntile(2).over(by_amount_unique).alias("ntile_2"),
).orderBy("region", "row_number").show()
+------+----+----------+------+----------+----+----------+--------+-------+
|region| rep| sale_date|amount|row_number|rank|dense_rank|pct_rank|ntile_2|
+------+----+----------+------+----------+----+----------+--------+-------+
| north|Asha|2026-01-02|   150|         1|   1|         1|     0.0|      1|
| north|Asha|2026-01-05|   150|         2|   1|         1|     0.0|      1|
| north| Ben|2026-01-01|   150|         3|   1|         1|     0.0|      1|
| north|Asha|2026-01-01|   100|         4|   4|         2|    0.75|      2|
| north| Ben|2026-01-03|    90|         5|   5|         3|     1.0|      2|
| south|Chen|2026-01-01|   200|         1|   1|         1|     0.0|      1|
| south|Dara|2026-01-04|   200|         2|   1|         1|     0.0|      1|
| south|Chen|2026-01-02|    50|         3|   3|         2|     1.0|      2|
+------+----+----------+------+----------+----+----------+--------+-------+
Function On ties Next value after a 3-way tie for first
row_number Different numbers, order among ties arbitrary unless you break ties 4
rank Same number 4 (gaps)
dense_rank Same number 2 (no gaps)

percent_rank is (rank - 1) / (rows in partition - 1), and ntile(n) splits each partition into n nearly equal buckets.

Top N per group

Compute the rank in a column, then filter on it. Window functions are not allowed directly in filter:

w1 = Window.partitionBy("rep").orderBy("sale_date")
try:
    sales.filter(F.row_number().over(w1) == 1).show()
except Exception as e:
    print(type(e).__name__, str(e).split(" claus")[0])
AnalysisException [WINDOW_FUNCTION_NOT_ALLOWED_IN_CLAUSE] It is not allowed to use window functions inside WHERE
w = Window.partitionBy("region").orderBy(F.desc("amount"), F.asc("rep"), F.asc("sale_date"))
top = sales.withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn")
top.orderBy("region").show()
+----+------+----------+------+
| rep|region| sale_date|amount|
+----+------+----------+------+
|Asha| north|2026-01-02|   150|
|Chen| south|2026-01-01|   200|
+----+------+----------+------+

The extra ordering columns (rep, sale_date) are tiebreakers. Without them, which of the three 150s gets rn = 1 depends on the physical order of rows after the shuffle and can change between runs, which makes pipelines non-reproducible and tests flaky.

If you want all tied rows, filter on rank() == 1 instead, or compare with a partition-wide maximum:

by_region = Window.partitionBy("region")
(sales.withColumn("best", F.max("amount").over(by_region))
      .filter(F.col("amount") == F.col("best")).drop("best")
      .orderBy("region", "rep", "sale_date").show())
+----+------+----------+------+
| rep|region| sale_date|amount|
+----+------+----------+------+
|Asha| north|2026-01-02|   150|
|Asha| north|2026-01-05|   150|
| Ben| north|2026-01-01|   150|
|Chen| south|2026-01-01|   200|
|Dara| south|2026-01-04|   200|
+----+------+----------+------+

Deduplicate: keep the latest record per key

The most common window in data engineering. A change feed delivers several versions of each customer, plus an exact duplicate; keep the newest version per key:

updates = spark.createDataFrame(
    [(1, "[email protected]", "2026-01-01 09:00:00"), (1, "[email protected]", "2026-01-03 10:00:00"),
     (2, "[email protected]", "2026-01-02 08:00:00"), (2, "[email protected]", "2026-01-02 08:00:00")],
    "customer_id INT, email STRING, updated_at STRING",
).withColumn("updated_at", F.to_timestamp("updated_at"))
latest_w = Window.partitionBy("customer_id").orderBy(F.desc("updated_at"), F.desc("email"))
latest = updates.withColumn("rn", F.row_number().over(latest_w)).filter("rn = 1").drop("rn")
latest.orderBy("customer_id").show()
+-----------+------------+-------------------+
|customer_id|       email|         updated_at|
+-----------+------------+-------------------+
|          1|[email protected]|2026-01-03 10:00:00|
|          2|   [email protected]|2026-01-02 08:00:00|
+-----------+------------+-------------------+

dropDuplicates(["customer_id"]) would also return one row per key, but it keeps an arbitrary one, not the latest. Use it only for exact duplicates.

Previous and next rows: lag and lead

by_rep = Window.partitionBy("rep").orderBy("sale_date")
changes = (sales
    .withColumn("previous", F.lag("amount").over(by_rep))
    .withColumn("change", F.col("amount") - F.col("previous"))
    .withColumn("next_date", F.lead("sale_date").over(by_rep))
    .withColumn("prev_or_0", F.lag("amount", 1, 0).over(by_rep)))
changes.orderBy("rep", "sale_date").show()
+----+------+----------+------+--------+------+----------+---------+
| rep|region| sale_date|amount|previous|change| next_date|prev_or_0|
+----+------+----------+------+--------+------+----------+---------+
|Asha| north|2026-01-01|   100|    NULL|  NULL|2026-01-02|        0|
|Asha| north|2026-01-02|   150|     100|    50|2026-01-05|      100|
|Asha| north|2026-01-05|   150|     150|     0|      NULL|      150|
| Ben| north|2026-01-01|   150|    NULL|  NULL|2026-01-03|        0|
| Ben| north|2026-01-03|    90|     150|   -60|      NULL|      150|
|Chen| south|2026-01-01|   200|    NULL|  NULL|2026-01-02|        0|
|Chen| south|2026-01-02|    50|     200|  -150|      NULL|      200|
|Dara| south|2026-01-04|   200|    NULL|  NULL|      NULL|        0|
+----+------+----------+------+--------+------+----------+---------+

lag(col, offset, default) looks offset rows back; the first row of each partition has nothing before it, so the result is NULL unless you pass a default. lead looks forward. Note that “previous row” is not “previous day”: Asha’s previous on 5 January is from 2 January. If you need calendar periods, build a complete date spine first, or use a range frame.

Running totals and the default frame

by_region_date = Window.partitionBy("region").orderBy("sale_date")
sales.select("region", "sale_date", "rep", "amount",
    F.sum("amount").over(by_region_date).alias("default_frame"),
    F.sum("amount").over(by_region_date.rowsBetween(Window.unboundedPreceding, Window.currentRow))
     .alias("rows_frame"),
    F.sum("amount").over(Window.partitionBy("region")).alias("partition_total"),
).orderBy("region", "sale_date", "rep").show()
+------+----------+----+------+-------------+----------+---------------+
|region| sale_date| rep|amount|default_frame|rows_frame|partition_total|
+------+----------+----+------+-------------+----------+---------------+
| north|2026-01-01|Asha|   100|          250|       100|            640|
| north|2026-01-01| Ben|   150|          250|       250|            640|
| north|2026-01-02|Asha|   150|          400|       400|            640|
| north|2026-01-03| Ben|    90|          490|       490|            640|
| north|2026-01-05|Asha|   150|          640|       640|            640|
| south|2026-01-01|Chen|   200|          200|       200|            450|
| south|2026-01-02|Chen|    50|          250|       250|            450|
| south|2026-01-04|Dara|   200|          450|       450|            450|
+------+----------+----+------+-------------+----------+---------------+

The frame rules explain the three columns:

  • With orderBy and no frame, Spark uses RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW. A range frame treats rows with the same ordering value as peers, so both rows on 1 January get 250. That surprises people who expect a row-by-row running total.
  • rowsBetween(unboundedPreceding, currentRow) counts physical rows, giving 100 then 250. (Which of two peers comes first is still arbitrary unless the order is unique.)
  • No orderBy means the frame is the whole partition: a partition total on every row.

Rolling windows: ROWS vs RANGE

rowsBetween(-1, 0) means “this row and the one before”, regardless of dates. rangeBetween(-2, 0) means “rows whose ordering value is within 2 of mine”, so it needs a single numeric ordering column. For dates, order by unix_date (days since epoch) or a timestamp cast to seconds:

last_3_days = (Window.partitionBy("rep").orderBy(F.unix_date("sale_date"))
               .rangeBetween(-2, 0))
last_2_rows = Window.partitionBy("rep").orderBy("sale_date").rowsBetween(-1, 0)
sales.select("rep", "sale_date", "amount",
    F.sum("amount").over(last_3_days).alias("sum_last_3_days"),
    F.avg("amount").over(last_2_rows).alias("avg_last_2_rows"),
).orderBy("rep", "sale_date").show()
+----+----------+------+---------------+---------------+
| rep| sale_date|amount|sum_last_3_days|avg_last_2_rows|
+----+----------+------+---------------+---------------+
|Asha|2026-01-01|   100|            100|          100.0|
|Asha|2026-01-02|   150|            250|          125.0|
|Asha|2026-01-05|   150|            150|          150.0|
| Ben|2026-01-01|   150|            150|          150.0|
| Ben|2026-01-03|    90|            240|          120.0|
|Chen|2026-01-01|   200|            200|          200.0|
|Chen|2026-01-02|    50|            250|          125.0|
|Dara|2026-01-04|   200|            200|          200.0|
+----+----------+------+---------------+---------------+

On 5 January, Asha’s 3-day sum is only 150, because her previous sale (2 January) is outside the window, while the 2-row average still includes it. For “last 7 days” metrics use a range frame; a rows frame silently spans weeks when data is sparse.

first, last and share of total

whole = (Window.partitionBy("rep").orderBy("sale_date")
         .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing))
sales.select("rep", "sale_date", "amount",
    F.first("amount").over(by_rep).alias("first_amount"),
    F.last("amount").over(by_rep).alias("last_default_frame"),
    F.last("amount").over(whole).alias("last_whole_partition"),
    F.round(F.col("amount") / F.sum("amount").over(Window.partitionBy("rep")), 2).alias("share"),
).orderBy("rep", "sale_date").show()
+----+----------+------+------------+------------------+--------------------+-----+
| rep| sale_date|amount|first_amount|last_default_frame|last_whole_partition|share|
+----+----------+------+------------+------------------+--------------------+-----+
|Asha|2026-01-01|   100|         100|               100|                 150| 0.25|
|Asha|2026-01-02|   150|         100|               150|                 150| 0.38|
|Asha|2026-01-05|   150|         100|               150|                 150| 0.38|
| Ben|2026-01-01|   150|         150|               150|                  90| 0.63|
| Ben|2026-01-03|    90|         150|                90|                  90| 0.38|
|Chen|2026-01-01|   200|         200|               200|                  50|  0.8|
|Chen|2026-01-02|    50|         200|                50|                  50|  0.2|
|Dara|2026-01-04|   200|         200|               200|                 200|  1.0|
+----+----------+------+------------+------------------+--------------------+-----+

last with the default frame returns the current row’s value, because the frame ends at the current row. To get the true last value of the partition, widen the frame to unboundedFollowing. This is one of the most common window bugs.

Forward-fill missing values

last(col, ignorenulls=True) over a frame ending at the current row carries the most recent non-NULL value forward, which is how you fill gaps in sensor readings or slowly changing attributes:

readings = spark.createDataFrame(
    [("s1", 1, 20.0), ("s1", 2, None), ("s1", 3, None), ("s1", 4, 23.0), ("s1", 5, None)],
    "sensor STRING, t INT, temp DOUBLE")
ff = Window.partitionBy("sensor").orderBy("t").rowsBetween(Window.unboundedPreceding, Window.currentRow)
readings.withColumn("filled", F.last("temp", ignorenulls=True).over(ff)).show()
+------+---+----+------+
|sensor|  t|temp|filled|
+------+---+----+------+
|    s1|  1|20.0|  20.0|
|    s1|  2|NULL|  20.0|
|    s1|  3|NULL|  20.0|
|    s1|  4|23.0|  23.0|
|    s1|  5|NULL|  23.0|
+------+---+----+------+

Sessions: gaps and islands

Split each user’s clicks into sessions whenever there is a gap of more than 30 minutes. Flag the start of each session with lag, then a running sum of the flags numbers the sessions:

clicks = spark.createDataFrame(
    [("u1", "2026-01-01 10:00:00"), ("u1", "2026-01-01 10:10:00"), ("u1", "2026-01-01 11:30:00"),
     ("u1", "2026-01-01 11:40:00"), ("u2", "2026-01-01 10:05:00")],
    "user STRING, ts STRING").withColumn("ts", F.to_timestamp("ts"))
u = Window.partitionBy("user").orderBy("ts")
sessions = (clicks
    .withColumn("gap_min", (F.col("ts").cast("long") - F.lag("ts").over(u).cast("long")) / 60)
    .withColumn("new_session", F.when(F.col("gap_min").isNull() | (F.col("gap_min") > 30), 1).otherwise(0))
    .withColumn("session_id", F.sum("new_session").over(u)))
sessions.orderBy("user", "ts").show()
+----+-------------------+-------+-----------+----------+
|user|                 ts|gap_min|new_session|session_id|
+----+-------------------+-------+-----------+----------+
|  u1|2026-01-01 10:00:00|   NULL|          1|         1|
|  u1|2026-01-01 10:10:00|   10.0|          0|         1|
|  u1|2026-01-01 11:30:00|   80.0|          1|         2|
|  u1|2026-01-01 11:40:00|   10.0|          0|         2|
|  u2|2026-01-01 10:05:00|   NULL|          1|         1|
+----+-------------------+-------+-----------+----------+

The same “flag, then running sum” pattern finds consecutive streaks and status changes. For streaming data, Spark also has built-in session_window grouping (covered in the streaming event-time lesson).

The same thing in Spark SQL

sales.createOrReplaceTempView("sales")
spark.sql("""
    SELECT rep, sale_date, amount,
           SUM(amount) OVER w AS running_total,
           ROW_NUMBER() OVER (PARTITION BY region ORDER BY amount DESC, rep, sale_date) AS rn
    FROM sales
    WINDOW w AS (PARTITION BY rep ORDER BY sale_date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)
    ORDER BY rep, sale_date
""").show()
+----+----------+------+-------------+---+
| rep| sale_date|amount|running_total| rn|
+----+----------+------+-------------+---+
|Asha|2026-01-01|   100|          100|  4|
|Asha|2026-01-02|   150|          250|  1|
|Asha|2026-01-05|   150|          400|  2|
| Ben|2026-01-01|   150|          150|  3|
| Ben|2026-01-03|    90|          240|  5|
|Chen|2026-01-01|   200|          200|  1|
|Chen|2026-01-02|    50|          250|  3|
|Dara|2026-01-04|   200|          200|  2|
+----+----------+------+-------------+---+

The WINDOW clause names a specification for reuse. DataFrame and SQL windows compile to the same plan.

How a window executes

def plan(df):
    buf = io.StringIO()
    with contextlib.redirect_stdout(buf):
        df.explain()
    text = re.sub(r"#\d+L?", "", buf.getvalue())
    print(re.sub(r", \[plan_id=\d+\]", "", text).strip())

plan(sales.withColumn("rn", F.row_number().over(Window.partitionBy("region").orderBy("amount"))))
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Window [row_number() windowspecdefinition(region, amount ASC NULLS FIRST, specifiedwindowframe(RowFrame, unboundedpreceding$(), currentrow$())) AS rn], [region], [amount ASC NULLS FIRST]
   +- Sort [region ASC NULLS FIRST, amount ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(region, 4), ENSURE_REQUIREMENTS
         +- Project [rep, region, cast(sale_date as date) AS sale_date, amount]
            +- Scan ExistingRDD[rep,region,sale_date,amount]

Bottom-up: a shuffle by the partition columns (Exchange hashpartitioning(region)), a sort within each partition by partition and order columns, then the Window operator walks each sorted partition. Each window partition (all rows for one region) is processed by one task, and its rows are buffered as the operator moves through them; very large window partitions can spill to disk.

Several columns over the same specification share one shuffle, sort and Window step:

w1 = Window.partitionBy("rep").orderBy("sale_date")
plan(sales.withColumn("a", F.lag("amount").over(w1)).withColumn("b", F.sum("amount").over(w1)))
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Window [lag(amount, -1, null) windowspecdefinition(rep, sale_date ASC NULLS FIRST, specifiedwindowframe(RowFrame, -1, -1)) AS a, sum(amount) windowspecdefinition(rep, sale_date ASC NULLS FIRST, specifiedwindowframe(RangeFrame, unboundedpreceding$(), currentrow$())) AS b], [rep], [sale_date ASC NULLS FIRST]
   +- Sort [rep ASC NULLS FIRST, sale_date ASC NULLS FIRST], false, 0
      +- Exchange hashpartitioning(rep, 4), ENSURE_REQUIREMENTS
         +- Project [rep, region, cast(sale_date as date) AS sale_date, amount]
            +- Scan ExistingRDD[rep,region,sale_date,amount]

The plan also confirms the default frame: sum got a RangeFrame from unbounded preceding to the current row. Windows with different partition columns each add their own shuffle and sort, so group window columns by specification.

Pitfalls

  1. No tiebreaker: row_number over non-unique ordering gives non-reproducible results.
  2. Default frame surprises: with orderBy, aggregates use a RANGE frame (peers share values) and last returns the current row.
  3. Filtering in the same step: compute the window column with withColumn, then filter.
  4. Skewed partitions: one partition key holding most rows makes one task do most of the work, and the operator buffers that partition. See partitions, shuffles and skew.
  5. No partitionBy on large data: everything goes through a single partition.
  6. NULLs in the order column sort first in ascending order by default; use asc_nulls_last() when NULL means “unknown, put at the end”.

In interviews

Window questions are some of the most frequent PySpark and SQL interview tasks: “top N per group”, “deduplicate keeping the latest record”, “month-over-month change”, “running total”, “7-day rolling average” and “sessionise clickstream events”. Strong answers:

  • pick row_number, rank or dense_rank based on how ties should behave, and add a tiebreaker;
  • state the frame explicitly and explain ROWS vs RANGE;
  • explain the cost: a shuffle by the partition columns and a sort, one task per window partition;
  • know the follow-up, “what happens without partitionBy?”: a single-partition shuffle.

Practice questions

Return the two highest-paid employees in each department, keeping ties. Which function do you use?

Use dense_rank() over Window.partitionBy("dept").orderBy(F.desc("salary")) and filter <= 2 if “two highest salaries” should include everyone on those salaries; use rank() if ties should consume places; use row_number() with a tiebreaker if you need exactly two rows per department.

Why does F.sum(“amount”).over(Window.partitionBy(“rep”).orderBy(“sale_date”)) give the same value on two rows with the same date?

With an orderBy and no explicit frame, Spark uses a RANGE frame from the start of the partition to the current row, and rows with equal ordering values are peers that share the frame. Use rowsBetween(Window.unboundedPreceding, Window.currentRow) (and a unique ordering) for a row-by-row running total.

How do you keep only the most recent record for each customer_id? Why not dropDuplicates?

Number rows with row_number() over partitionBy("customer_id").orderBy(F.desc("updated_at"), <tiebreaker>) and keep rn = 1. dropDuplicates(["customer_id"]) keeps an arbitrary row per key, not the latest.

F.last(“amount”).over(Window.partitionBy(“rep”).orderBy(“sale_date”)) returns the current row’s amount. Why?

The default frame with an ordering ends at the current row, so the last value in the frame is the current row (or its last peer). Use rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing) to see the whole partition.

Compute a 7-day rolling sum per store when some days have no sales. Rows or range frame?

A range frame on a numeric day value, for example orderBy(F.unix_date("day")).rangeBetween(-6, 0), so the window covers 7 calendar days regardless of missing rows. A rowsBetween(-6, 0) frame covers the last 7 rows, which can span many more days.

What does the physical plan of a window look like, and what is expensive about it?

An Exchange hashpartitioning on the partition columns, a Sort on partition and order columns, then a Window operator. The shuffle and sort are the cost; each window partition is processed by a single task, so skewed keys create stragglers, and a window with no partitionBy uses Exchange SinglePartition.

Key takeaways

  • A window keeps every row and computes over a partition, an order and a frame; set all three deliberately.
  • row_number needs a tiebreaker to be deterministic; rank and dense_rank keep ties, with and without gaps.
  • With orderBy and no frame, aggregates use a RANGE frame to the current row, and last returns the current row.
  • Use rowsBetween for “N rows” and rangeBetween on a numeric ordering for “N days”; sparse data makes them differ.
  • Deduplication, forward-fill and sessionisation are all window patterns worth practising.
  • Windows cost a shuffle by the partition columns plus a sort; reuse specifications, watch for skew, and never leave out partitionBy on large data.

By DataDank Editorial · Last reviewed Oct 2026 · All examples run on PySpark 4.2.0 in local mode (local[2], spark.sql.shuffle.partitions=4); plans are printed with expression IDs removed. The window API is the same in Spark 3.x.

Progress is saved in this browser only. No account needed.

Sources and further reading

Search
Filter by type