scala - 删除接近时间戳的条目

Question

我想删除所有重复条目的记录，但已经说过时间戳的差异可以是任何时间量作为偏移量，但为了简单起见，将使用 2 分钟。

+-------------------+-----+----+
|Date               |ColA |ColB|
+-------------------+-----+----+
|2017-07-04 18:50:21|ABC  |DEF |
|2017-07-04 18:50:26|ABC  |DEF |
|2017-07-04 18:50:21|ABC  |KLM |
+-------------------+-----+----+

我希望我的数据框只有行

+-------------------+-----+----+
|Date               |ColA |ColB|
+-------------------+-----+----+
|2017-07-04 18:50:26|ABC  |DEF |
|2017-07-04 18:50:21|ABC  |KLM |
+-------------------+-----+----+

我尝试了类似的方法，但这并没有删除重复项。

    val joinedDfNoDuplicates = joinedDFTransformed.as("df1").join(joinedDFTransformed.as("df2"), col("df1.ColA") === col("df2.ColA") &&
      col("df1.ColB") === col("df2.ColB") && 
      && abs(unix_timestamp(col("Date")) - unix_timestamp(col("Date"))) > offset
      )

现在，我只是在这里选择不同的或按分钟分组 Find minimum for a timestamp through Spark groupBy dataframe基于某些列的数据，但我想要一个更强大的解决方案，原因是该间隔之外的数据可能是有效数据。此外，偏移量可以更改，因此可能会根据要求在 5 秒或 5 分钟内更改。

有人向我提到创建一个 UDF 来比较日期以及所有其他列是否相同，但我不确定具体如何执行此操作，这样我要么过滤掉行，要么添加一个标志，然后删除这些行以获取任何帮助将不胜感激。

类似的sql问题在这里Duplicate entries with different timestamp

谢谢!

Answer 1

我会这样做:

1. 定义一个窗口来对虚拟列上的日期进行排序。
2. 添加一个虚拟列，并为其添加一个常量值。
3. 添加一个新列，其中包含上一条记录的日期。
4. 计算当前日期与前一个日期之间的差异。
5. 根据差值过滤您的记录。

代码可以如下所示:

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

val w = Window.partitionBy("dummy").orderBy("Date") // step 1

df.withColumn("dummy", lit(1)) // this is step 1
  .withColumn("previousDate", lag($"Date", 1) over w) // step 2
  .withColumn("difference", unix_timestamp($"Date") - unix_timestamp("previousDate")) // step 3

如果您有时间可能接近的记录对，则上述解决方案有效。如果您有两条以上的记录，则可以将每条记录与窗口中的第一条记录(而不是前一条记录)进行比较，因此您可以使用第一个($"日期")。在本例中，“差异”列包含当前记录与窗口中第一条记录之间的时间差。