激发 UDF 未正确提供滚动计数

Question

我有一个Spark UDF来计算列的滚动计数，精确地说是wt时间。如果我需要计算24小时的滚动计数，例如以时间2020-10-02 09:04:00进入，我需要回看直到2020-10-01 09:04:00（非常精确）。

如果我在本地运行，滚动计数UDF工作良好，并给出正确的计数，但当我在集群上运行时，它给出的结果不正确。下面是示例输入和输出

输入

+---------+-----------------------+
|OrderName|Time                   |
+---------+-----------------------+
|a        |2020-07-11 23:58:45.538|
|a        |2020-07-12 00:00:07.307|
|a        |2020-07-12 00:01:08.817|
|a        |2020-07-12 00:02:15.675|
|a        |2020-07-12 00:05:48.277|
+---------+-----------------------+

预期产出

+---------+-----------------------+-----+
|OrderName|Time                   |Count|
+---------+-----------------------+-----+
|a        |2020-07-11 23:58:45.538|1    |
|a        |2020-07-12 00:00:07.307|2    |
|a        |2020-07-12 00:01:08.817|3    |
|a        |2020-07-12 00:02:15.675|1    |
|a        |2020-07-12 00:05:48.277|1    |
+---------+-----------------------+-----+

最后两个条目值在本地是4和5，但是在集群上它们是不正确的。我最大的猜测是数据分布在执行器之间，udf也在每个执行器上被并行调用。由于UDF的参数之一是列(在这个例子中是分区键-顺序名)，如果是这样的话，我该如何控制/纠正集群的行为呢？以便以正确的方式为每个分区计算正确的计数。有什么建议吗

Answer 1

根据您的评论，您想要统计过去24小时内每条记录的记录总数

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.types.LongType

//A sample data (Guessing from your question)
val df = Seq(("a","2020-07-10 23:58:45.438","1"),("a","2020-07-11 23:58:45.538","1"),("a","2020-07-11 23:58:45.638","1")).toDF("OrderName","Time","Count")

// Extract the UNIX TIMESTAMP for your time column
val df2 = df.withColumn("unix_time",concat(unix_timestamp($"Time"),split($"Time","\\.")(1)).cast(LongType))

val noOfMilisecondsDay : Long = 24*60*60*1000

//Create a window per `OrderName` and select rows from `current time - 24 hours` to `current time` 
val winSpec = Window.partitionBy("OrderName").orderBy("unix_time").rangeBetween(Window.currentRow - noOfMilisecondsDay, Window.currentRow)

// Final you perform your COUNT or SUM(COUNT) as per your need
val finalDf = df2.withColumn("tot_count", count("OrderName").over(winSpec))

//or val finalDf = df2.withColumn("tot_count", sum("Count").over(winSpec))