SQL window functions for data engineers: running totals, lag and QUALIFY

Most of a data engineer's SQL is not clever. It counts, adds up and picks rows, every day, for tables other people trust. This module teaches that SQL on 1 small shop: 8 orders and a customer system that records each change to a customer. This part covers grouping and window functions, the tool for running totals, rankings and "the newest row per key". The next parts design the tables a warehouse keeps, a star schemaStar SchemaA way to arrange warehouse tables for reporting: a fact table of events and their numbers, joined by keys to dimension tables that describe them, such as customer, product and date. with a customer table that remembers the past.

Run SQL on Databricks or your laptop

On Databricks Free Edition, make a notebook and start each cell with %sql on its own first line, or set the notebook's language to SQL. Paste each statement below into its own cell.

On your laptop, with PySpark installed as in module 2, save this as run_sql.py. It runs every statement of the files you give it in 1 Spark session and shows what each query returns:

"""Run each statement of 1 or more .sql files in 1 Spark session, and show what every query returns."""
import re
import sys

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("sql").enableHiveSupport().getOrCreate()
spark.sparkContext.setLogLevel("ERROR")

for path in sys.argv[1:]:
    text = open(path).read()
    text = re.sub(r"--[^\n]*", "", text)  # drop comments
    for statement in filter(None, (part.strip() for part in text.split(";"))):
        result = spark.sql(statement)
        if re.match(r"(?i)(select|with)\b", statement):
            result.show(truncate=False)

enableHiveSupport() keeps a list of your tables in a metastore_db folder, so the tables the last part makes are still there on the next run, as they are on Databricks.

Then save the shop's data as setup.sql. Temporary views live as long as the session, which is all a lesson needs:

SET TIME ZONE 'UTC';

CREATE OR REPLACE TEMP VIEW orders AS
SELECT CAST(order_id AS BIGINT) AS order_id, customer_id, ordered_at,
       CAST(amount AS DECIMAL(10,2)) AS amount
FROM VALUES
  (1001, 'C01', TIMESTAMP'2026-10-01 09:15:00', 25.50),
  (1002, 'C02', TIMESTAMP'2026-10-01 10:02:00', 12.00),
  (1003, 'C03', TIMESTAMP'2026-10-02 11:30:00', 60.00),
  (1004, 'C01', TIMESTAMP'2026-10-02 12:45:00', 40.00),
  (1005, 'C02', TIMESTAMP'2026-10-02 16:20:00', 18.25),
  (1006, 'C01', TIMESTAMP'2026-10-04 08:05:00', 30.00),
  (1007, 'C03', TIMESTAMP'2026-10-04 13:40:00', 15.00),
  (1008, 'C02', TIMESTAMP'2026-10-05 19:10:00', 22.75)
AS t(order_id, customer_id, ordered_at, amount);

CREATE OR REPLACE TEMP VIEW customer_changes AS
SELECT * FROM VALUES
  ('C01', 'Ana',  'PT', TIMESTAMP'2026-09-01 08:00:00'),
  ('C02', 'Ben',  'IE', TIMESTAMP'2026-09-15 10:00:00'),
  ('C03', 'Chen', 'DE', TIMESTAMP'2026-10-02 12:00:00'),
  ('C01', 'Ana',  'ES', TIMESTAMP'2026-10-03 09:30:00')
AS t(customer_id, name, country, changed_at);

The casts give each column the type a real table would have: written as 25.50, a number is a decimal(4,2), which holds nothing above 99.99. Setting the time zone keeps every date the same wherever you run it. Put the queries of this part in aggregates.sql and run python run_sql.py setup.sql aggregates.sql.

Group, then filter the groups

GROUP BY returns 1 row per group. Revenue per day:

SELECT to_date(ordered_at) AS order_day,
       count(*) AS orders,
       count(DISTINCT customer_id) AS customers,
       sum(amount) AS revenue
FROM orders
GROUP BY to_date(ordered_at)
ORDER BY order_day;

+----------+------+---------+-------+
|order_day |orders|customers|revenue|
+----------+------+---------+-------+
|2026-10-01|2     |2        |37.50  |
|2026-10-02|3     |3        |118.25 |
|2026-10-04|2     |2        |45.00  |
|2026-10-05|1     |1        |22.75  |
+----------+------+---------+-------+

2026-10-03 is missing, because no order was placed that day. A report that needs every day joins to a table of dates, which part 2 builds.

WHERE filters rows before they are grouped; HAVING filters the groups after. The customers with at least 3 orders:

SELECT customer_id, count(*) AS orders, sum(amount) AS spent
FROM orders
GROUP BY customer_id
HAVING count(*) >= 3
ORDER BY customer_id;

+-----------+------+-----+
|customer_id|orders|spent|
+-----------+------+-----+
|C01        |3     |95.50|
|C02        |3     |53.00|
+-----------+------+-----+

A window keeps every row

A window function computes a value for each row from a group of rows around it, and keeps the row. OVER (...) describes the group: PARTITION BY splits the rows, ORDER BY sorts each part, and a frame says which of the sorted rows count. The running total of revenue, written 2 ways:

SELECT order_id, to_date(ordered_at) AS order_day, amount,
       sum(amount) OVER (ORDER BY to_date(ordered_at)) AS running_default,
       sum(amount) OVER (ORDER BY to_date(ordered_at), order_id
                         ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running_rows
FROM orders
ORDER BY order_id;

+--------+----------+------+---------------+------------+
|order_id|order_day |amount|running_default|running_rows|
+--------+----------+------+---------------+------------+
|1001    |2026-10-01|25.50 |37.50          |25.50       |
|1002    |2026-10-01|12.00 |37.50          |37.50       |
|1003    |2026-10-02|60.00 |155.75         |97.50       |
|1004    |2026-10-02|40.00 |155.75         |137.50      |
|1005    |2026-10-02|18.25 |155.75         |155.75      |
|1006    |2026-10-04|30.00 |200.75         |185.75      |
|1007    |2026-10-04|15.00 |200.75         |200.75      |
|1008    |2026-10-05|22.75 |223.50         |223.50      |
+--------+----------+------+---------------+------------+

The 2 columns agree only at the end of each day. With ORDER BY and no frame written, Spark uses a range frame: every row whose value is up to the current row's, so all 3 orders of 2 October count for each of them. That is a running total per day, which may be what you want, but nobody reading the query can tell it is what you meant. ROWS counts rows instead, and order_id in the ORDER BY fixes which of 2 orders on the same day comes first, so the query returns the same numbers on every run.

Press the orders in the demo to see which rows each function reads.

SELECT order_id, sum(amount) OVER (ORDER BY to_date(ordered_at)) AS result
FROM orders

Number rows, and look back 1 row

row_number() counts the rows of each window from 1. lag() reads a column from the row before, here to measure the days since each customer's previous order:

SELECT customer_id, order_id, to_date(ordered_at) AS order_day,
       row_number() OVER (PARTITION BY customer_id ORDER BY ordered_at) AS nth,
       datediff(to_date(ordered_at),
                to_date(lag(ordered_at) OVER (PARTITION BY customer_id ORDER BY ordered_at))) AS days_since_last
FROM orders
ORDER BY customer_id, ordered_at;

+-----------+--------+----------+---+---------------+
|customer_id|order_id|order_day |nth|days_since_last|
+-----------+--------+----------+---+---------------+
|C01        |1001    |2026-10-01|1  |NULL           |
|C01        |1004    |2026-10-02|2  |1              |
|C01        |1006    |2026-10-04|3  |2              |
|C02        |1002    |2026-10-01|1  |NULL           |
|C02        |1005    |2026-10-02|2  |1              |
|C02        |1008    |2026-10-05|3  |3              |
|C03        |1003    |2026-10-02|1  |NULL           |
|C03        |1007    |2026-10-04|2  |2              |
+-----------+--------+----------+---+---------------+

Each customer's first order has no row before it, so lag() returns NULL, and the difference is NULL too. PARTITION BY customer_id is what keeps C01's orders from reading C02's.

Keep the top row of each group with QUALIFY

QUALIFY filters on a window function, as HAVING filters on an aggregate. Each customer's largest order, with no subquery:

SELECT customer_id, order_id, amount
FROM orders
QUALIFY row_number() OVER (PARTITION BY customer_id ORDER BY amount DESC) = 1
ORDER BY customer_id;

+-----------+--------+------+
|customer_id|order_id|amount|
+-----------+--------+------+
|C01        |1004    |40.00 |
|C02        |1008    |22.75 |
|C03        |1003    |60.00 |
+-----------+--------+------+

The same pattern, ordered by the time a row changed, keeps the newest version of each key, as module 2 did in PySpark. If 2 orders tie on the ORDER BY, row_number() still keeps 1 of them, and which one is not fixed; add a column that breaks the tie.

A window can also run over the result of GROUP BY, because Spark groups first and computes windows after. rank() gives tied rows the same place and skips the next:

SELECT customer_id, sum(amount) AS spent,
       rank() OVER (ORDER BY sum(amount) DESC) AS place
FROM orders
GROUP BY customer_id
ORDER BY place;

+-----------+-----+-----+
|customer_id|spent|place|
+-----------+-----+-----+
|C01        |95.50|1    |
|C03        |75.00|2    |
|C02        |53.00|3    |
+-----------+-----+-----+

These queries answer questions about orders. The next part asks how to keep orders and customers in tables that answer every such question the same way: facts, dimensions and the grain of each.