Deduplicação

Na tabela eventos, o e2 chegou duas vezes. Na camada bronze isso fica como chegou; a silver é onde ele vira um só.

SELECT id_evento, COUNT(*) AS vezes
FROM eventos
GROUP BY id_evento
HAVING COUNT(*) > 1;

id_evento | vezes
--------- | -----
e2        | 2

Cópia exata: DISTINCT

SELECT COUNT(*) AS brutos,
       (SELECT COUNT(*) FROM (SELECT DISTINCT * FROM eventos)) AS distintos
FROM eventos;

brutos | distintos
------ | ---------
7      | 6

Mesma chave, versões diferentes: ROW_NUMBER

Quando a "cópia" muda em algum campo (o status do pedido foi atualizado), DISTINCT não ajuda. A regra é de negócio: aqui, fica o último evento de cada pedido.

WITH numerados AS (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY pedido_id ORDER BY ts DESC, id_evento DESC) AS n
  FROM eventos
)
SELECT pedido_id, tipo AS ultimo_estado, ts
FROM numerados
WHERE n = 1
ORDER BY pedido_id;

pedido_id | ultimo_estado | ts
--------- | ------------- | -------------------
101       | pago          | 2024-06-01 09:05:00
102       | pago          | 2024-06-01 10:30:00
103       | cancelado     | 2024-06-02 11:20:00

O desempate (id_evento DESC) não é enfeite: sem ele, dois eventos com o mesmo ts fazem a escolha depender da ordem física dos dados, e o resultado muda entre execuções.

No PySpark

from pyspark.sql import Window, functions as F

eventos = spark.table("bronze.eventos")
unicos = eventos.dropDuplicates(["id_evento"])

ultimo = Window.partitionBy("pedido_id").orderBy(F.desc("ts"), F.desc("id_evento"))
estado_atual = (unicos
    .withColumn("n", F.row_number().over(ultimo))
    .where("n = 1")
    .drop("n"))

• dropDuplicates(["id_evento"]) mantém uma linha por chave, sem dizer qual. Se importa qual, use a janela.
• Em streaming, dropDuplicates guarda estado para sempre; use withWatermark para limitar quanto tempo um evento repetido ainda é reconhecido.