Deduplicating and Ranking Data with PySpark Window Functions

A window keeps the row in view

A groupBy("region").agg(...) produces one result row per region. A window can calculate over a region and attach the result to each order row. Watch the comparison in How Do Spark Window Functions Work? A Practical Guide to PySpark Window Functions by DecisionForest.

A window specification answers two questions:

• partitionBy: Which rows belong to the same logical group?
• orderBy: In what order should those rows be considered?

For example, Window.partitionBy("region").orderBy("event_day") treats each region separately and establishes an order within it. A window partition is a logical group for the calculation; do not assume it already matches the way input files or Spark tasks are partitioned.

The Spark SQL reference expresses the same ideas as PARTITION BY and ORDER BY inside OVER (...). Read its description and frame definition before coding.

We will use three versions of an order feed throughout. Each row has an event_id, the order’s region and day, an update timestamp, an ingestion sequence number, and an amount in integer cents.

from pyspark.sql import Window
from pyspark.sql import functions as F

raw = spark.createDataFrame(
    [
        ("o1", "East", "2025-01-01", "2025-01-03 08:00:00", 1, 1000),
        ("o1", "East", "2025-01-01", "2025-01-04 08:00:00", 2, 1300),
        ("o2", "East", "2025-01-02", "2025-01-03 09:00:00", 3, 2000),
        ("o3", "East", "2025-01-02", "2025-01-03 09:00:00", 4, 2000),
        ("o4", "East", "2025-01-03", "2025-01-03 10:00:00", 5,  500),
        ("o5", "West", "2025-01-01", "2025-01-03 10:00:00", 6,  600),
        ("o5", "West", "2025-01-01", "2025-01-03 10:00:00", 7,  800),
        ("o6", "West", "2025-01-02", "2025-01-04 10:00:00", 8,  800),
        ("o7", "West", "2025-01-02", "2025-01-04 11:00:00", 9, 1200),
    ],
    "event_id STRING, region STRING, event_day STRING, "
    "updated_at STRING, ingest_seq LONG, amount_cents LONG"
)

orders = raw.select(
    "event_id",
    "region",
    F.to_date("event_day").alias("event_day"),
    F.to_timestamp("updated_at").alias("updated_at"),
    "ingest_seq",
    "amount_cents"
)

The sample uses parseable date and timestamp strings. A later lesson will address explicit schemas and malformed or missing values; here, the focus is on choosing the right rows and window ordering.

Deduplicate by defining what “latest” means

Suppose the business rule is one current row per event_id. Choose the row with the latest updated_at; if two versions have the same timestamp, choose the greater ingest_seq. This is more specific than merely “remove duplicates.” For o1, the 1,300-cent version should win. For o5, the timestamps tie, so the 800-cent version should win.

Define that rule as an ordered window, number the rows, then retain number one:

latest_version = (
    Window.partitionBy("event_id")
          .orderBy(
              F.col("updated_at").desc_nulls_last(),
              F.col("ingest_seq").desc()
          )
)

latest = (
    orders
    .withColumn("version_number", F.row_number().over(latest_version))
    .filter(F.col("version_number") == 1)
    .drop("version_number")
)

row_number() starts at one within each event_id window. The filter then changes the dataset from all received versions to one selected version per order. Unlike groupBy, the window first retains every row so it can assign a position; the subsequent filter is what removes rows.

A reliable deduplication rule needs a deterministic winner. This example assumes ingest_seq distinguishes versions whose update timestamps tie. If both ordering fields can tie, add a stable unique source identifier or define another business rule. Otherwise, row_number() still assigns a one, but which tied row receives it is not guaranteed. Likewise, if IDs are only unique within a source system, the deduplication key must include that source—not just event_id.

Why not dropDuplicates(["event_id"])? It expresses that only one row should remain, but not which version should remain. That is insufficient when updates can change the amount or region.

Rank rows without losing ties

Now that each order has one current version, suppose an interviewer asks for “the highest-value orders in each region.” The word rank needs clarification: should equally valued orders share a position, and should the next position have a gap?

Watch the tie examples in 54. row_number(), rank(), dense_rank() functions in PySpark by WafaStudies. Concentrate on the results for tied values, rather than copying notebook syntax from the demonstration.

For our orders, there are two useful orderings. To give every row a predictable individual position, break amount ties with event_id. To give equal amounts the same rank, order by amount alone:

individual_order = (
    Window.partitionBy("region")
          .orderBy(F.col("amount_cents").desc(), F.col("event_id").asc())
)

amount_order = (
    Window.partitionBy("region")
          .orderBy(F.col("amount_cents").desc())
)

ranked = (
    latest
    .withColumn("row_number", F.row_number().over(individual_order))
    .withColumn("rank", F.rank().over(amount_order))
    .withColumn("dense_rank", F.dense_rank().over(amount_order))
)

ranked.select(
    "region", "event_id", "amount_cents",
    "row_number", "rank", "dense_rank"
).orderBy("region", "row_number").show()

The expected rankings are:

The functions answer different questions. row_number() assigns a distinct position to each row; it suits “keep exactly one” or “return exactly two orders per region.” rank() gives tied values the same position and leaves a gap afterward: East’s next order after two orders ranked first is ranked third. dense_rank() also preserves ties but does not leave a gap: East’s next distinct amount is ranked second.

Notice why event_id appears only in individual_order. If it were added to amount_order, the two 2,000-cent East orders would no longer have identical ordering keys, defeating the intended tie in rank() and dense_rank(). When answering a “top N” question, state whether N means exactly N rows or all rows in the top N ranked positions.

Calculate a cumulative result with an explicit frame

Ranking asks where a row stands relative to others. A running total asks which earlier rows should contribute to the value attached to the current row.

DecisionForest’s next segment shows an important trap: with an ordered aggregate window, equal ordering values can cause peers to receive the same cumulative result. Watch the change when an explicit row-based frame is added.

For a running order-revenue total, specify both a stable order and a ROWS frame from the first row through the current row:

running_window = (
    Window.partitionBy("region")
          .orderBy(F.col("event_day").asc(), F.col("event_id").asc())
          .rowsBetween(Window.unboundedPreceding, Window.currentRow)
)

running = latest.withColumn(
    "running_revenue_cents",
    F.sum("amount_cents").over(running_window)
)

running.select(
    "region", "event_day", "event_id",
    "amount_cents", "running_revenue_cents"
).orderBy("region", "event_day", "event_id").show()

The expected cumulative results are:

The total restarts for West because region defines the window partition. Within East, o2 and o3 share a day, but event_id establishes their sequence. ROWS then includes exactly the rows up to each row’s position. Without that tie-breaker, which same-day order receives 3,300 rather than 5,300 would be unstable.

An ordered aggregate window without an explicit ROWS frame can use range-based behavior that treats tied ordering values as peers. For a per-record running total, writing the frame explicitly makes the intended behavior clear. If the requirement instead says “one cumulative figure per day,” you would first decide how to combine same-day orders; that is a different output grain.

Explain the cost and the rule

In an interview, describe the business rule before the function name: “I partition by order ID and order by update timestamp plus a stable tie-breaker to choose the latest version.” For ranking, explain whether ties share a position. For a running total, name both the ordering columns and the frame.

Windows also connect to the execution concepts from earlier lessons. Spark may need to redistribute rows so records with the same window key can be processed together, then sort them within that key. On a large feed, a window over a heavily concentrated key can be expensive. None of the withColumn calls runs the computation immediately; an action such as the final show() triggers execution.

Conclusion

You can now use row_number() with a deterministic ordering to deduplicate updates, choose among row_number(), rank(), and dense_rank() based on tie semantics, and use sum(...).over(...) with an explicit row frame for cumulative results. The central habit is to define which rows belong together, their exact order, and whether ties are meaningful before writing the window.

Next, you will make transformations more robust by enforcing schemas and handling null, malformed, and nested input records.