如何根据火花结构化流中的时间戳字段进行重复数据删除和保持最新？

Question

Spark dropDuplicates 保留第一个实例并忽略该键的所有后续出现。是否可以在保留最近出现的情况下删除重复项？

例如，如果下面是我获得的微批次，那么我想保留每个国家/地区的最新记录（按时间戳字段排序）。

batchId：0

Australia, 10, 2020-05-05 00:00:06
Belarus, 10, 2020-05-05 00:00:06

batchId：1

Australia, 10, 2020-05-05 00:00:08
Belarus, 10, 2020-05-05 00:00:03

那么batchId 1之后的输出应该低于-

Australia, 10, 2020-05-05 00:00:08
Belarus, 10, 2020-05-05 00:00:06

更新 1 这是我拥有的当前代码

//KafkaDF is a streaming dataframe created from Kafka as source
val streamingDF = kafkaDF.dropDuplicates("country")

streamingDF.writeStream
    .trigger(Trigger.ProcessingTime(10000L))
    .outputMode("update")
    .foreachBatch {
      (batchDF: DataFrame, batchId: Long) => {
        println("batchId: "+ batchId)
        batchDF.show()
      }
    }.start()

我想输出所有新行或时间戳大于迄今为止处理的先前批次中的任何记录的所有行。下面的例子

batchId: 0 之后 - 这两个国家都是第一次出现，所以我应该让它们输出

Australia, 10, 2020-05-05 00:00:06
Belarus, 10, 2020-05-05 00:00:06

batchId: 1 之后 - 白俄罗斯的时间戳比我们在第 0 批中收到的时间戳要旧，所以我不会在输出中显示它。显示澳大利亚是因为它的时间戳比我目前看到的更新。

Australia, 10, 2020-05-05 00:00:08

现在假设 batchId 2 将两条记录都作为迟到的记录出现，那么它不应该在该批次的输出中显示任何内容。

输入batchId：2

Australia, 10, 2020-05-05 00:00:01
Belarus, 10, 2020-05-05 00:00:01

batchId: 2 之后

.

Update-2

为每个批次添加输入和预期记录。用红色标记的行将被丢弃，并且不会在输出中显示为具有相同国家名称的另一行，并且在之前的批次中可以看到更近的时间戳

Answer 1

为了避免流式应用程序中迟到的事件，您需要在应用程序中保持一个状态，在您的情况下跟踪每个键的最新处理事件，它是country。

case class AppState(country:String, latestTs:java.sql.Timestamp)

对于微批处理，您可能会收到多个事件，当您执行groupByKey(_.country) 时，您将获得属于key(country) 的事件，您需要将其与状态进行比较以找到最新的输入事件并更新使用密钥的最新时间戳声明并继续处理最新事件以进行进一步处理。对于迟到的事件，它应该返回一个Option[Event]，并在后续过程中过滤掉它。

详细说明请参考blog。

Answer 2

尝试在 Spark Streaming 中使用window 函数，如下所示。

val columns = Seq("country","id").map(col(_))
df.groupBy(window($"timestamp","10 minutes","5 minutes"), columns:_*)

您也可以检查相同的in this question，解决方案在python中。