Window Functions

A window is a set of rows related to the current row. Unlike groupBy, a window function keeps every row — it just adds a column computed over the window. You define the window with Window.partitionBy(...).orderBy(...), then apply a function over it. row_number is the classic: number the rows 1, 2, 3… within each partition.

import org.apache.spark.sql.expressions.Window
val w = Window.partitionBy($"category").orderBy($"amount".desc)
def row_number(): Column         // 1,2,3 … no ties, always distinct
someFn.over(w)                   // apply a function over the window

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.row_number

val w = Window.partitionBy($"category").orderBy($"amount".desc)

orders.select(
  $"product", $"category", $"amount",
  row_number().over(w).as("rn")
).show(false)

Three ranking functions that differ only in how they handle ties. row_number breaks ties arbitrarily; rank gives ties the same number then skips (1,2,2,4); dense_rank gives ties the same number but doesn't skip (1,2,2,3). percent_rank expresses the rank as a 0–1 fraction.

def rank(): Column          // 1,2,2,4  — gaps after ties
def dense_rank(): Column    // 1,2,2,3  — no gaps
def percent_rank(): Column  // (rank-1)/(n-1), between 0.0 and 1.0

import org.apache.spark.sql.functions.{rank, dense_rank}

val w = Window.orderBy($"qty".desc)

orders.select(
  $"product", $"qty",
  rank().over(w).as("rank"),
  dense_rank().over(w).as("dense_rank")
).show(false)

lag pulls a value from a previous row in the window; lead from a following row. This is how you compute deltas — "how much bigger was this order than the customer's last one?" — without a self-join.

def lag(e: Column, offset: Int = 1): Column
def lag(e: Column, offset: Int, defaultValue: Any): Column
def lead(e: Column, offset: Int = 1): Column

import org.apache.spark.sql.functions.{lag, lead}

val w = Window.partitionBy($"user_id").orderBy($"ordered_at")

orders.select(
  $"user_id", $"ordered_at", $"amount",
  lag($"amount").over(w).as("prev_amt"),
  lead($"amount").over(w).as("next_amt")
).show(false)

Apply a plain aggregate (sum, avg) over an ordered window and it becomes cumulative: each row sees all rows up to and including itself. Add an explicit frame with rowsBetween to control exactly which rows are included — the whole basis of running totals and moving averages.

Window.orderBy($"ordered_at").rowsBetween(Window.unboundedPreceding, Window.currentRow)
// rowsBetween(-2, 0)  → current row + the 2 before it (a 3-row moving window)
sum($"amount").over(w)   avg($"amount").over(w)

import org.apache.spark.sql.functions.sum

val w = Window.orderBy($"ordered_at")
  .rowsBetween(Window.unboundedPreceding, Window.currentRow)

orders.select(
  $"ordered_at", $"amount",
  sum($"amount").over(w).as("running_total")
).show(false)

Grab a specific row's value from within the window: the first_value, the last_value, or the nth_value. Great for "what was the top product in this category?" stamped onto every row.

def first_value(e: Column): Column      def last_value(e: Column): Column
def nth_value(e: Column, offset: Int): Column
// last_value usually needs an explicit full frame — see the note

import org.apache.spark.sql.functions.first_value

val w = Window.partitionBy($"category").orderBy($"amount".desc)

orders.select(
  $"product", $"category", $"amount",
  first_value($"product").over(w).as("top_product")
).show(false)

ntile(n) splits the ordered window into n roughly-equal buckets and labels each row 1…n — the way you build quartiles, deciles, or A/B cohorts. cume_dist gives the cumulative distribution (fraction of rows ≤ this one).

def ntile(n: Int): Column       // bucket number 1..n
def cume_dist(): Column         // cumulative distribution, 0 < x <= 1

import org.apache.spark.sql.functions.ntile

val w = Window.orderBy($"amount".desc)

orders.select(
  $"product", $"amount",
  ntile(2).over(w).as("half")
).show(false)