pysparkintermediate45 minPySpark

Multi-Level Window Ranking & Deduplication

Deduplicate at-least-once distributed payment event streams by partitioning by merchant ID and ordering by timestamp with row_number().

Start buildingOpen related lesson

The incident

Company
FinPay Gateway
Engineer
Neha Sharma, Data Engineering Lead
Trigger
Duplicate payment webhook retries created duplicate payment status records in downstream merchant settlement tables.
Stakes
Over 40 lakh rupees in double-counted transaction settlements threatened banking reconciliation ledgers.

On a Monday morning, Neha Sharma arrived at FinPay Gateway, a financial technology platform processing payment settlements between merchants and 12 banking partners. The automated reconciliation ledger had flagged a severe anomaly: daily settlement totals were inflated by more than 40 lakh rupees. Partner merchants reported seeing duplicate transactions on their merchant portals, and banking clearing partners were refusing to sign off on morning settlement files. Reconciling payment ledgers required immediate elimination of phantom duplicate charges before banking clearing windows closed. Any uncorrected discrepancy would halt outgoing fund disbursements and trigger statutory banking penalty inquiries.

Neha investigated the incoming event pipeline. FinPay webhook ingestion was designed with at-least-once delivery guarantees. During network timeouts or intermittent gateway hiccups, upstream payment switches resent webhook events. Because the downstream batch pipeline did not perform deterministic deduplication, retried events with identical merchant identifiers and payment IDs were appended into analytical tables as distinct rows. In distributed streaming and messaging architectures, duplicate deliveries are an inevitable operational reality that pipelines must actively manage through idempotent transformation stages.

The initial engineering response was to call dropDuplicates(['merchant_id']). However, this naive approach caused severe data corruption: dropDuplicates in Spark retains an arbitrary record when partitioning keys without respect to event timestamps. Consequently, outdated 'pending' transaction statuses were retained while authoritative 'completed' statuses were dropped, creating false reconciliation exceptions. Discarding the latest event state distorted merchant settlement balances and generated phantom dispute alerts across customer support queues. Several prominent enterprise retailers reported incorrect settlement receipts within hours.

Neha redesigned the deduplication logic using PySpark distributed window functions. By defining an explicit Window partition across merchant_id and ordering chronologically by event timestamp descending, the pipeline ranks occurrences per merchant deterministically using row_number(). Filtering on rank equal to 1 guarantees that only the latest authoritative event state is retained. Window specifications provide fine-grained control over record selection.

Neha implemented the window ranking pipeline in PySpark. Her code defined the Window partition, assigned row numbers, filtered for top-ranked records, and cleaned auxiliary ranking metadata. When executed against duplicated test streams, the pipeline correctly retained the latest completed status for merchant 101 (850.00 rupees) and merchant 102 (1,450.00 rupees), cleanly eliminating 40 lakh rupees in phantom duplicate volume. The pipeline verified that total authoritative volume equaled exactly 2,620.00 rupees across the test merchant cohort without dropping valid transactions.

Neha presented the verified deduplication architecture to the Chief Risk Officer and the Banking Settlement Manager. The deterministic window ranking pattern eliminated duplicate settlements across all 12 banking partners. The window ranking framework was adopted as the mandatory deduplication standard for all FinPay streaming and micro-batch ingestion pipelines, safeguarding partner trust and protecting the platform from double-crediting liabilities as transaction throughput scaled across national banking corridors.

What you will learn

  • Define distributed analytical windows using Window.partitionBy() and orderBy()
  • Assign deterministic chronological rankings per entity using F.row_number()
  • Isolate the latest authoritative record state by filtering on row_number == 1
  • Differentiate deterministic window ranking from arbitrary dropDuplicates() behavior

What you need first

Window Specifications in PySparkPySpark window functions

Window.partitionBy(key).orderBy(col) defines the logical slice of data across which ranking functions evaluate.

PythonMini example
from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql import functions as F
spark = SparkSession.builder.master('local[1]').getOrCreate()
w = Window.partitionBy('k').orderBy('v')
df = spark.createDataFrame([(1, 10)], ['k', 'v'])
print(df.withColumn('r', F.row_number().over(w)).first()['r'])
1
Watch out

Forgetting to specify an orderBy clause inside Window, causing row_number() to fail or behave non-deterministically.

Row Number vs Dense RankAggregations & grouping

row_number() assigns strictly unique integers (1, 2, 3) even when ordering keys are tied, preventing duplicate retention.

PythonMini example
from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql import functions as F
spark = SparkSession.builder.master('local[1]').getOrCreate()
w = Window.partitionBy('k').orderBy('v')
df = spark.createDataFrame([(1, 10), (1, 10)], ['k', 'v'])
print([r['rn'] for r in df.withColumn('rn', F.row_number().over(w)).collect()])
[1, 2]
Watch out

Using rank() or dense_rank() for deduplication, which retains multiple records when timestamps are tied.

Architecture and data flow

Duplicate Webhook Event Ingestion
Duplicate Webhook Event Ingestion. Network retries deliver multiple status updates per transaction, inflating settlement ledgers.
Window Partitioning and Ranking DAG
Window Partitioning and Ranking DAG. Events are partitioned by entity key, ranked chronologically, and filtered on top rank.
Authoritative Deduplicated Event Mart
Authoritative Deduplicated Event Mart. The output mart retains strictly the single newest event per merchant.

The data

A payment status event stream containing duplicate network retry deliveries.

payment_events: Payment status change events with network retries.

event_idmerchant_idamountstatustimestamp
e1101850.0pending2024-03-01 10:00:00
e2101850.0completed2024-03-01 10:00:05
e31021450.0pending2024-03-01 10:01:00
e41021450.0completed2024-03-01 10:01:10
e5103320.0completed2024-03-01 10:02:00

Step by step

Read each step, run its code, then check your output against the one shown.

  1. Initialize payment event stream DataFrame with duplicate states

    We initialize the payment events DataFrame containing sample webhook events across merchants 101, 102, and 103. Notice that merchant 101 has two records: an initial 'pending' event followed by a 'completed' event five seconds later. Similarly, merchant 102 has two events. In at-least-once distributed messaging systems, network retries and event updates produce multiple rows for the same entity key, necessitating rigorous deduplication before loading downstream financial ledgers. We display all 5 raw event rows to observe chronological lifecycle progressions across each merchant partition.

    PythonStep 1
    from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[1]").appName("window_ranking").getOrCreate()
events = [
    ("e1", 101, 850.0, "pending", "2024-03-01 10:00:00"),
    ("e2", 101, 850.0, "completed", "2024-03-01 10:00:05"),
    ("e3", 102, 1450.0, "pending", "2024-03-01 10:01:00"),
    ("e4", 102, 1450.0, "completed", "2024-03-01 10:01:10"),
    ("e5", 103, 320.0, "completed", "2024-03-01 10:02:00"),
]
df = spark.createDataFrame(events, ["event_id", "merchant_id", "amount", "status", "timestamp"])
df.show(truncate=False)
    +--------+-----------+------+---------+-------------------+
|event_id|merchant_id|amount|status   |timestamp          |
+--------+-----------+------+---------+-------------------+
|e1      |101        |850.0 |pending  |2024-03-01 10:00:00|
|e2      |101        |850.0 |completed|2024-03-01 10:00:05|
|e3      |102        |1450.0|pending  |2024-03-01 10:01:00|
|e4      |102        |1450.0|completed|2024-03-01 10:01:10|
|e5      |103        |320.0 |completed|2024-03-01 10:02:00|
+--------+-----------+------+---------+-------------------+
    Common mistake

    Assuming raw event streams contain strictly unique entity keys without verifying ingestion guarantees.

    How you will know

    You will see 5 payment events displayed with event_id, merchant_id, amount, status, and timestamp.

  2. Demonstrate data loss with arbitrary dropDuplicates deduplication

    Wrong turn first: this step shows the mistake before the fix

    Here is the naive approach developers often attempt first: calling df.dropDuplicates(['merchant_id']). In Apache Spark, dropDuplicates without sorting picks an arbitrary row from each partition. When run on our dataset, it arbitrarily keeps the outdated 'pending' state for merchant 101, discarding the final 'completed' event. In financial systems, dropping the latest authoritative state leads to severe reconciliation mismatches and customer disputes. We demonstrate this flawed attempt to show why non-deterministic deduplication methods corrupt event lifecycle histories and distort financial ledger reporting.

    PythonStep 2
    from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[1]").appName("window_ranking").getOrCreate()
events = [
    ("e1", 101, 850.0, "pending", "2024-03-01 10:00:00"),
    ("e2", 101, 850.0, "completed", "2024-03-01 10:00:05"),
    ("e3", 102, 1450.0, "pending", "2024-03-01 10:01:00"),
    ("e4", 102, 1450.0, "completed", "2024-03-01 10:01:10"),
    ("e5", 103, 320.0, "completed", "2024-03-01 10:02:00"),
]
df = spark.createDataFrame(events, ["event_id", "merchant_id", "amount", "status", "timestamp"])

# Naive approach: arbitrary deduplication
naive_dedup = df.dropDuplicates(["merchant_id"])
print("Naive deduplication results (arbitrary state retained):")
naive_dedup.select("merchant_id", "status", "timestamp").show(truncate=False)
    Naive deduplication results (arbitrary state retained):
+-----------+---------+-------------------+
|merchant_id|status   |timestamp          |
+-----------+---------+-------------------+
|101        |pending  |2024-03-01 10:00:00|
|102        |pending  |2024-03-01 10:01:00|
|103        |completed|2024-03-01 10:02:00|
+-----------+---------+-------------------+
    Common mistake

    Using dropDuplicates on streaming event logs without chronological ordering, discarding newer records.

    How you will know

    Observe that naive deduplication retains whatever row Spark encounters first without timestamp awareness.

  3. Construct Window specification partitioned by merchant_id

    We build a deterministic window specification using the Window utility from pyspark.sql.window. We partition the dataset by merchant_id and order rows within each partition by timestamp descending. Partitioning by merchant_id groups all lifecycle events for a given merchant into the same logical evaluation window, while ordering by descending timestamp places the most recent event at rank 1. This constructs the analytical foundation for deterministic state resolution.

    PythonStep 3
    from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[1]").appName("window_ranking").getOrCreate()
events = [
    ("e1", 101, 850.0, "pending", "2024-03-01 10:00:00"),
    ("e2", 101, 850.0, "completed", "2024-03-01 10:00:05"),
    ("e3", 102, 1450.0, "pending", "2024-03-01 10:01:00"),
    ("e4", 102, 1450.0, "completed", "2024-03-01 10:01:10"),
    ("e5", 103, 320.0, "completed", "2024-03-01 10:02:00"),
]
df = spark.createDataFrame(events, ["event_id", "merchant_id", "amount", "status", "timestamp"])

window_spec = Window.partitionBy("merchant_id").orderBy(F.col("timestamp").desc())
df.select("merchant_id", "status", "timestamp").orderBy("merchant_id", F.col("timestamp").desc()).show(truncate=False)
    +-----------+---------+-------------------+
|merchant_id|status   |timestamp          |
+-----------+---------+-------------------+
|101        |completed|2024-03-01 10:00:05|
|101        |pending  |2024-03-01 10:00:00|
|102        |completed|2024-03-01 10:01:10|
|102        |pending  |2024-03-01 10:01:00|
|103        |completed|2024-03-01 10:02:00|
+-----------+---------+-------------------+
    Common mistake

    Ordering timestamps in ascending order instead of descending, which assigns rank 1 to the oldest record.

    How you will know

    Output shows events ordered by timestamp descending within each merchant partition.

  4. Compute row_number ranking over analytical window

    We evaluate the analytical window by applying F.row_number().over(window_spec) and aliasing the result to 'rn'. Within each merchant partition, PySpark assigns strictly consecutive integers starting at 1 for the newest event. For merchant 101, the 'completed' event (10:00:05) receives rank 1 while 'pending' (10:00:00) receives rank 2. We display the ranked events DataFrame to verify that ranking strictly follows chronological recency. Explicit row numbering ensures that ties are resolved deterministically without creating duplicate ranking assignments across distributed executor partitions or distorting event state.

    PythonStep 4
    from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[1]").appName("window_ranking").getOrCreate()
events = [
    ("e1", 101, 850.0, "pending", "2024-03-01 10:00:00"),
    ("e2", 101, 850.0, "completed", "2024-03-01 10:00:05"),
    ("e3", 102, 1450.0, "pending", "2024-03-01 10:01:00"),
    ("e4", 102, 1450.0, "completed", "2024-03-01 10:01:10"),
    ("e5", 103, 320.0, "completed", "2024-03-01 10:02:00"),
]
df = spark.createDataFrame(events, ["event_id", "merchant_id", "amount", "status", "timestamp"])

window_spec = Window.partitionBy("merchant_id").orderBy(F.col("timestamp").desc())
ranked_df = df.withColumn("rn", F.row_number().over(window_spec))
ranked_df.select("merchant_id", "status", "timestamp", "rn").show(truncate=False)
    +-----------+---------+-------------------+---+
|merchant_id|status   |timestamp          |rn |
+-----------+---------+-------------------+---+
|101        |completed|2024-03-01 10:00:05|1  |
|101        |pending  |2024-03-01 10:00:00|2  |
|102        |completed|2024-03-01 10:01:10|1  |
|102        |pending  |2024-03-01 10:01:00|2  |
|103        |completed|2024-03-01 10:02:00|1  |
+-----------+---------+-------------------+---+
    Common mistake

    Using F.rank() instead of F.row_number(), which assigns tied rank 1 to records with identical timestamps.

    How you will know

    Output shows an rn column where the latest event per merchant has rn=1.

  5. Filter for latest authoritative state and drop ranking metadata

    In the final step, we filter the ranked DataFrame on F.col('rn') == 1 to retain strictly the latest event per merchant, and drop the auxiliary 'rn' column. The resulting DataFrame contains exactly three clean records: merchant 101 in 'completed' status (850.00), merchant 102 in 'completed' status (1,450.00), and merchant 103 in 'completed' status (320.00). Total deduplicated revenue equals exactly 2,620.00 rupees, cleanly eliminating phantom duplicate payments while preserving authoritative transaction amounts and accurate lifecycle states across all active merchant ledgers in the financial database.

    PythonStep 5
    from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[1]").appName("window_ranking").getOrCreate()
events = [
    ("e1", 101, 850.0, "pending", "2024-03-01 10:00:00"),
    ("e2", 101, 850.0, "completed", "2024-03-01 10:00:05"),
    ("e3", 102, 1450.0, "pending", "2024-03-01 10:01:00"),
    ("e4", 102, 1450.0, "completed", "2024-03-01 10:01:10"),
    ("e5", 103, 320.0, "completed", "2024-03-01 10:02:00"),
]
df = spark.createDataFrame(events, ["event_id", "merchant_id", "amount", "status", "timestamp"])

window_spec = Window.partitionBy("merchant_id").orderBy(F.col("timestamp").desc())
deduped = (
    df
    .withColumn("rn", F.row_number().over(window_spec))
    .filter(F.col("rn") == 1)
    .drop("rn")
    .orderBy("merchant_id")
)
deduped.show(truncate=False)
total = deduped.agg(F.sum("amount")).first()[0]
print(f"Total Authoritative Volume: {total:.2f}")
    +--------+-----------+------+---------+-------------------+
|event_id|merchant_id|amount|status   |timestamp          |
+--------+-----------+------+---------+-------------------+
|e2      |101        |850.0 |completed|2024-03-01 10:00:05|
|e4      |102        |1450.0|completed|2024-03-01 10:01:10|
|e5      |103        |320.0 |completed|2024-03-01 10:02:00|
+--------+-----------+------+---------+-------------------+

Total Authoritative Volume: 2620.00
    Common mistake

    Leaving the auxiliary ranking column in production schemas, leaking internal transformation artifacts.

    How you will know

    Output displays exactly 3 deduplicated rows with status completed and total volume 2620.00.

Build it yourself

Write your solution below, run it in your browser, and pass every check.

PySpark (Catalyst Engine)Ready to run
Loading editor…
Press Run Code to evaluate
Script executes cleanly without PySpark syntax or analysis errors
Uses row_number() window function for chronological deduplication
Retains authoritative completed status for all merchants
Calculates correct authoritative volume of 2620.00

What the result means

Interpretation

Window ranking provides deterministic state resolution for at-least-once distributed event feeds. By partitioning on merchant_id and sorting on descending timestamp, row_number() == 1 guarantees that only the newest authoritative record is retained, eliminating 40 lakh rupees in phantom duplicate transactions.

The message you would send

Merchant ledger deduplication complete: we resolved event stream duplicate retries using PySpark window ranking. All merchants now reflect their authoritative latest completed transaction states. The 40 lakh rupee phantom settlement discrepancy has been eliminated.

Put it on your CV

Resume or LinkedIn bullet

OutputCopy this line
Engineered deterministic streaming deduplication pipelines in PySpark using window ranking functions to eliminate duplicate ledger settlements.

README for GitHub

OutputREADME.md
# Multi-Level Window Ranking & Deduplication in PySpark

## Problem
At-least-once webhook deliveries duplicated payment events, inflating settlement ledgers by over 40 lakh rupees.

## Solution Architecture
- Partitioned event streams by merchant_id.
- Ordered by timestamp descending.
- Assigned F.row_number().over(window_spec) and filtered rn == 1.

## Results
Guaranteed deterministic latest-state deduplication and restored 100% reconciliation accuracy across 12 partner banks.
Stretch goal

Extend the window specification to handle millisecond-level timestamp ties using event_id UUIDs.

Keep going