Running totals

Accumulate values with explicit row frames and partition-specific resets.

Step 1 of 6 · Learn

Accumulate through the current row

sum over an ordered window adds values from the start through the current row. rowsBetween(unboundedPreceding, currentRow) makes that frame explicit. The example processes order_id ascending. Reversing the window order changes which rows have already contributed, even if the final table remains sorted ascending. A global window is appropriate here only because the teaching dataset is small.

Lesson reference: PySpark and Scala

Accumulate through the current row

PySpark example

from pyspark.sql import Window
from pyspark.sql.functions import col, sum

result = orders.withColumn(
    "running_total", sum("total").over(
        Window.orderBy(col("order_id").asc())
        .rowsBetween(Window.unboundedPreceding, Window.currentRow)
    )
).select("order_id", "status", "customer_id", "total", "running_total").orderBy("order_id")
result.show(truncate=False)

Scala example

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

val result = orders.withColumn(
    "running_total", sum("total").over(
        Window.orderBy(col("order_id").asc())
        .rowsBetween(Window.unboundedPreceding, Window.currentRow)
    )
).select("order_id", "status", "customer_id", "total", "running_total").orderBy("order_id")
result.show(false)

Restart for each group

partitionBy(status) keeps a separate accumulator for each status. Interleaved rows of other statuses do not contribute. Within each group, order_id determines the sequence and the explicit row frame includes the current row. Changing the partition key changes where totals restart.

PySpark example

from pyspark.sql import Window
from pyspark.sql.functions import col, sum

result = orders.withColumn(
    "running_total", sum("total").over(
        Window.partitionBy("status").orderBy(col("order_id").asc())
        .rowsBetween(Window.unboundedPreceding, Window.currentRow)
    )
).select("order_id", "status", "customer_id", "total", "running_total").orderBy("order_id")
result.show(truncate=False)

Scala example

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

val result = orders.withColumn(
    "running_total", sum("total").over(
        Window.partitionBy("status").orderBy(col("order_id").asc())
        .rowsBetween(Window.unboundedPreceding, Window.currentRow)
    )
).select("order_id", "status", "customer_id", "total", "running_total").orderBy("order_id")
result.show(false)

Count qualifying rows so far

count(*) over the same row frame counts every row encountered. A conditional sum can count only matching rows without removing the other rows from the result. Use 1 for matches and 0 otherwise. Filtering the DataFrame first would remove rows that should remain visible.

PySpark example

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

result = orders.withColumn(
    "running_count", count("*").over(
        Window.orderBy("order_id").rowsBetween(Window.unboundedPreceding, Window.currentRow)
    )
).select("order_id", "status", "running_count").orderBy("order_id")
result.show(truncate=False)

Scala example

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

val result = orders.withColumn(
    "running_count", count("*").over(
        Window.orderBy("order_id").rowsBetween(Window.unboundedPreceding, Window.currentRow)
    )
).select("order_id", "status", "running_count").orderBy("order_id")
result.show(false)

example.pyPySpark

from pyspark.sql import Window from pyspark.sql.functions import col, sum result = orders.withColumn( "running_total", sum("total").over( Window.orderBy(col("order_id").asc()) .rowsBetween(Window.unboundedPreceding, Window.currentRow) ) ).select("order_id", "status", "customer_id", "total", "running_total").orderBy("order_id") result.show(truncate=False)

Ordered rowsGrow the row frameAccumulate values

Window sequenceOrder IDs in window order1001 → 1002 → 1003 → 1004 → 1005 → 1006

Source orders6 rows

order_id	customer_id	status	total	item_count
1001	101	Delivered	89.5	2
1002	102	Shipped	149	3
1003	101	Cancelled	35	1
1004	103	Delivered	219.99	4
1005	104	Delivered	49.99	1
1006	105	Shipped	120	2

Accumulate through the current row

Result6 rows

order_id	status	customer_id	total	running_total
1001	Delivered	101	89.5	89.5
1002	Shipped	102	149	238.5
1003	Cancelled	101	35	273.5
1004	Delivered	103	219.99	493.49
1005	Delivered	104	49.99	543.48
1006	Shipped	105	120	663.48

Order 1002 includes 1001 and 1002: 89.50 + 149.00 = 238.50. The final row includes all six orders.

The example is loaded in the editor. Run it as written, then try a small change.

Runs on the Spark backend. First startup may take a moment.

Run your code to see the result.