(Пример монетизации рекламы для блоков данных) Как найти последнее совпадение в потоке?

Question

В блоге "Представление объединений Stream-Stream в Apache Spark 2.3" обсуждается объединение кликов с показами на основе их adId:

# Define watermarks
impressionsWithWatermark = impressions \
  .selectExpr("adId AS impressionAdId", "impressionTime") \
  .withWatermark("impressionTime", "10 seconds ")   # max 10 seconds late

clicksWithWatermark = clicks \
  .selectExpr("adId AS clickAdId", "clickTime") \
  .withWatermark("clickTime", "20 seconds")        # max 20 seconds late

# Inner join with time range conditions
impressionsWithWatermark.join(
  clicksWithWatermark,
  expr(""" 
   clickAdId = impressionAdId AND 
    clickTime >= impressionTime AND 
    clickTime <= impressionTime + interval 1 minutes    
    """
  )
)

Я хотел бы знать, возможно ли отфильтровать результирующий поток так, чтобы в каждый "интервал запроса" были включены только строки с последним значением clickTime.

Интервал запроса - это интервал, указанный в условии соединения запроса:

clickTime >= impressionTime AND 
clickTime <= impressionTime + interval 1 minutes

Таким образом, я мог бы получить следующую последовательность

{type:impression, impressionAdId:1, timestamp: 1}
{type:click, clickAdId:1, timestamp: 1}
{type:click, clickAdId:1, timestamp: 15}

И после t=60 с или около того искра испускает следующую строку в кадре данных:

{impressionTimestamp: 1, clickTimestamp: 15: clickAddId: 1, impressionAdId: 1}

Я разместил только код на Python, потому что это было в статье, также приветствуются ответы с кодом Java или Scala.