Given a dataframe, how would you keep only the latest record for each customer?

You can use a window function to rank records per customer by a timestamp column and then filter the top row. For example:

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

val dfWithRank = df
  .withColumn("rn", row_number().over(Window.partitionBy("customer_id").orderBy(desc("event_time"))))

val latestPerCustomer = dfWithRank.filter(col("rn") === 1).drop("rn")

This partitions the data by customer_id, orders each partition by event_time descending, assigns a row number, and keeps only the first row per customer. Alternatively, if you only need the maximum timestamp per customer, you can use groupBy("customer_id").agg(max("event_time").alias("max_time")) and join back to the original dataframe to filter the matching rows. Both approaches run in O(n) time and are efficient for large datasets.