Gaps and islands

The pattern

An island is a run of consecutive rows that belong together: days in a login streak, events in one session, readings while a machine was in the same status. A gap is where one island ends and the next begins. Many interview questions are this pattern in disguise:

Flag, then running sum

One method solves all of them, in three window steps:

1. 1
   Look back. Use lag over a window partitioned by the entity and ordered by time, to put the previous row's value next to each row.
2. 2
   Flag the starts. Set new_run to 1 where a new island begins (no previous row, or the gap rule is met), else 0.
3. 3
   Number the islands. A running sum(new_run) over the same window gives every row its island number: it only goes up at a start.

Then group by the entity and island number to get each island's start, end and length.

Run the example

from pyspark.sql import functions as F
from pyspark.sql.window import Window
w = Window.partitionBy("athlete").orderBy("week")
running = w.rowsBetween(Window.unboundedPreceding, Window.currentRow)
flagged = (workouts
    .withColumn("prev_week", F.lag("week").over(w))
    .withColumn("new_run", F.when(F.col("prev_week").isNull() | (F.col("week") - F.col("prev_week") > 1), 1).otherwise(0))
    .withColumn("run_id", F.sum("new_run").over(running)))
result = (flagged.groupBy("athlete", "run_id")
    .agg(F.min("week").alias("first_week"), F.max("week").alias("last_week"), F.count("*").alias("weeks"))
    .orderBy("athlete", "run_id"))

WITH prev AS (
  SELECT athlete, week, LAG(week) OVER (PARTITION BY athlete ORDER BY week) AS prev_week
  FROM workouts
), flagged AS (
  SELECT athlete, week,
         CASE WHEN prev_week IS NULL OR week - prev_week > 1 THEN 1 ELSE 0 END AS new_run
  FROM prev
), numbered AS (
  SELECT athlete, week,
         SUM(new_run) OVER (PARTITION BY athlete ORDER BY week
                            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS run_id
  FROM flagged
)
SELECT athlete, run_id, MIN(week) AS first_week, MAX(week) AS last_week, COUNT(*) AS weeks
FROM numbered
GROUP BY athlete, run_id
ORDER BY athlete, run_id

Ana trained three weeks in a row (weeks 1 to 3), then two; Bo's longest run is four weeks. For "the longest run per athlete", aggregate once more with max("weeks").

The row_number shortcut

For consecutive integers or dates, there is a well-known trick: inside a run, week and row_number() both go up by 1, so their difference stays constant. That difference identifies the island.

from pyspark.sql import functions as F
from pyspark.sql.window import Window
w = Window.partitionBy("athlete").orderBy("week")
result = (workouts
    .withColumn("block", F.col("week") - F.row_number().over(w))
    .groupBy("athlete", "block")
    .agg(F.min("week").alias("first_week"), F.count("*").alias("weeks"))
    .filter(F.col("weeks") >= 3)
    .orderBy("athlete", "first_week"))

SELECT athlete, MIN(week) AS first_week, COUNT(*) AS weeks
FROM (SELECT athlete, week, week - ROW_NUMBER() OVER (PARTITION BY athlete ORDER BY week) AS block
      FROM workouts)
GROUP BY athlete, block
HAVING COUNT(*) >= 3
ORDER BY athlete, first_week

The same method, other questions

What it costs Optional deep dive

All windows here share the same partitionBy and orderBy, so Spark sorts once and evaluates them together: one shuffleshuffle: Moving rows between machines so that all rows with the same key end up together. Needed by joins, groupBy and sorting, and usually the most expensive step of a job. Learn more → by user, one sort, then the group by. A skewedskew: When one key has far more rows than the others, so one task does most of the work while the rest wait. Learn more → entity (one bot with millions of events) puts all its rows in one tasktask: The work for one partition in one stage, run on one CPU core. Learn more →, so check for hot keys first.

Common mistakes

• ✗Not deduplicating before the row_number trick — Two rows on the same day make the difference change inside a streak.
• ✗Forgetting the first row — lag is null on the first row of each partitionpartition: A chunk of a DataFrame's rows. Spark processes each partition as one task, so partitions decide how much work runs in parallel. Learn more →; it must start an island, or it gets run_id 0 and merges wrongly.
• ✗Using the default frame for the running sum — With ties in the order column, the default range frame sums tied rows together. Use rowsBetween.
• ✗Comparing timestamps as strings — Convert to seconds (unix_timestamp) before subtracting.