Deduplication

In the events table, e2 arrived twice. In the bronze layer this stays as it arrived; silver is where it becomes one.

SELECT event_id, COUNT(*) AS times
FROM events
GROUP BY event_id
HAVING COUNT(*) > 1;

event_id | times
-------- | -----
e2       | 2

Exact copy: DISTINCT

SELECT COUNT(*) AS raw,
       (SELECT COUNT(*) FROM (SELECT DISTINCT * FROM events)) AS distinct_rows
FROM events;

raw | distinct_rows
--- | -------------
7   | 6

Same key, different versions: ROW_NUMBER

When the "copy" changes in some field (the order status was updated), DISTINCT does not help. The rule is a business rule: here, the last event of each order is kept.

WITH numbered AS (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY ts DESC, event_id DESC) AS n
  FROM events
)
SELECT order_id, type AS latest_state, ts
FROM numbered
WHERE n = 1
ORDER BY order_id;

order_id | latest_state | ts
-------- | ------------ | -------------------
101      | paid         | 2024-06-01 09:05:00
102      | paid         | 2024-06-01 10:30:00
103      | canceled     | 2024-06-02 11:20:00

The tie-breaker (event_id DESC) is not decoration: without it, two events with the same ts make the choice depend on the physical order of the data, and the result changes between runs.

In PySpark

from pyspark.sql import Window, functions as F

events = spark.table("bronze.events")
unique = events.dropDuplicates(["event_id"])

latest = Window.partitionBy("order_id").orderBy(F.desc("ts"), F.desc("event_id"))
current_state = (unique
    .withColumn("n", F.row_number().over(latest))
    .where("n = 1")
    .drop("n"))

• dropDuplicates(["event_id"]) keeps one row per key, without saying which. If which one matters, use the window.
• In streaming, dropDuplicates keeps state forever; use withWatermark to limit how long a repeated event is still recognized.