Data Engineer Coding Practice

WITH ranked AS (
  SELECT *, ROW_NUMBER() OVER (
    PARTITION BY user_id ORDER BY event_at DESC
  ) AS rn
  FROM events
)
SELECT * FROM ranked WHERE rn = 1;

def validate(records, schema):
    clean, bad = [], []
    for i, r in enumerate(records):
        errs = check(r, schema)
        (bad if errs else clean).append(
            {"i": i, "errs": errs, "r": r}
            if errs else r)
    return clean, bad

w = Window.partitionBy("user_id") \
          .orderBy(F.desc("event_at"))
df.withColumn("rn", F.row_number().over(w)) \
  .filter("rn = 1") \
  .drop("rn")

[Source] -> [Kafka topic] -> [Flink job]
              ↓
        [DLQ topic]
                       ↓
                  [S3 bronze]
                       ↓
                 [dbt incremental]
                       ↓
               [Snowflake gold]