删除关闭时间戳的条目答案

【问题标题】：Dropping entries of close timestamps删除关闭时间戳的条目
【发布时间】：2017-10-26 17:45:12
【问题描述】：

我想删除所有重复条目的记录，但我说时间戳的差异可能是任何时间量的偏移量，但为简单起见将使用 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

谢谢！

【问题讨论】：

标签： scala apache-spark spark-dataframe

【解决方案1】：

我会这样做：

定义一个窗口以在一个虚拟列上排序日期。
添加一个虚拟列，并为其添加一个常量值。
添加一个包含上一条记录日期的新列。
计算日期与前一个日期之间的差异。
根据差值过滤您的记录。

代码可以如下所示：

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

如果您有可能及时接近的记录对，则上述解决方案是有效的。如果您有两条以上的记录，您可以将每条记录与窗口中的第一条记录（不是前一条）进行比较，因此您可以使用lag($"Date",1)，而不是使用lag($"Date",1)。在这种情况下，“差异”列包含当前记录与窗口中第一条记录之间的时间差。

【讨论】：