火花窗口函数之间的范围产生不正确的结果

Question

我试图在Spark DataFrame上使用RangeBetween对Long类型的列执行窗口函数，但窗口的结果不正确。我做错什么了吗？

val rowsRdd: RDD[Row] = spark.sparkContext.parallelize(
      Seq(
        Row("2014-11-01 08:10:10.12345", 141482941012345L),
        Row("2014-11-01 09:10:10.12345", 141483301012345L),
        Row("2014-11-01 10:10:10.12345", 141483661012345L),
        Row("2014-11-02 10:10:10.12345", 141492301012345L),
        Row("2014-11-03 10:10:10.12345", 141500941012345L),
        Row("2014-11-04 10:10:10.12345", 141509581012345L),
        Row("2014-11-05 10:10:10.12345", 141518221012345L),
        Row("2014-11-06 10:10:10.12345", 141526861012345L),
        Row("2014-11-07 10:10:10.12345", 141535501012345L),
        Row("2014-11-08 10:10:10.12345", 141544141012345L)
      )
    )
val schema = new StructType()
  .add(StructField("dateTime", StringType, true))
  .add(StructField("unixTime", LongType, true))

val df = spark.createDataFrame(rowsRdd, schema)
df.show(10, false)
df.printSchema()

+-------------------------+---------------+
|dateTime                 |unixTime       |
+-------------------------+---------------+
|2014-11-01 08:10:10.12345|141482941012345|
|2014-11-01 09:10:10.12345|141483301012345|
|2014-11-01 10:10:10.12345|141483661012345|
|2014-11-02 10:10:10.12345|141492301012345|
|2014-11-03 10:10:10.12345|141500941012345|
|2014-11-04 10:10:10.12345|141509581012345|
|2014-11-05 10:10:10.12345|141518221012345|
|2014-11-06 10:10:10.12345|141526861012345|
|2014-11-07 10:10:10.12345|141535501012345|
|2014-11-08 10:10:10.12345|141544141012345|
+-------------------------+---------------+

root
 |-- dateTime: string (nullable = true)
 |-- unixTime: long (nullable = true)

第一列是事件的时间戳（字符串，我们在实践中不会使用它），第二列是时间戳对应的unix时间，单位为10E-5秒。

现在，我想计算当前行的窗口中的事件数。例如，在3小时窗口中，我做：

val hour: Long = 60*60*100000L
val w = Window.orderBy(col("unixTime")).rangeBetween(-3*hour, 0)
val df2 = df.withColumn("cts", count(col("dateTime")).over(w)).orderBy(asc("unixTime"))

正确返回：

+-------------------------+---------------+---+
|dateTime                 |unixTime       |cts|
+-------------------------+---------------+---+
|2014-11-01 08:10:10.12345|141482941012345|1  |
|2014-11-01 09:10:10.12345|141483301012345|2  |
|2014-11-01 10:10:10.12345|141483661012345|3  |
|2014-11-02 10:10:10.12345|141492301012345|1  |
|2014-11-03 10:10:10.12345|141500941012345|1  |
|2014-11-04 10:10:10.12345|141509581012345|1  |
|2014-11-05 10:10:10.12345|141518221012345|1  |
|2014-11-06 10:10:10.12345|141526861012345|1  |
|2014-11-07 10:10:10.12345|141535501012345|1  |
|2014-11-08 10:10:10.12345|141544141012345|1  |
+-------------------------+---------------+---+

val hour: Long = 60*60*100000L
val w = Window.orderBy(col("unixTime")).rangeBetween(-6*hour, 0)
val df2 = df.withColumn("cts", count(col("dateTime")).over(w)).orderBy(asc("unixTime"))

+-------------------------+---------------+---+
|dateTime                 |unixTime       |cts|
+-------------------------+---------------+---+
|2014-11-01 08:10:10.12345|141482941012345|0  |
|2014-11-01 09:10:10.12345|141483301012345|0  |
|2014-11-01 10:10:10.12345|141483661012345|0  |
|2014-11-02 10:10:10.12345|141492301012345|0  |
|2014-11-03 10:10:10.12345|141500941012345|0  |
|2014-11-04 10:10:10.12345|141509581012345|0  |
|2014-11-05 10:10:10.12345|141518221012345|0  |
|2014-11-06 10:10:10.12345|141526861012345|0  |
|2014-11-07 10:10:10.12345|141535501012345|0  |
|2014-11-08 10:10:10.12345|141544141012345|0  |
+-------------------------+---------------+---+

val hour: Long = 60*60*100000L
val w = Window.orderBy(col("unixTime")).rangeBetween(-12*hour, 0)
val df2 = df.withColumn("cts", count(col("dateTime")).over(w)).orderBy(asc("unixTime"))

+-------------------------+---------------+---+
|dateTime                 |unixTime       |cts|
+-------------------------+---------------+---+
|2014-11-01 08:10:10.12345|141482941012345|1  |
|2014-11-01 09:10:10.12345|141483301012345|1  |
|2014-11-01 10:10:10.12345|141483661012345|1  |
|2014-11-02 10:10:10.12345|141492301012345|1  |
|2014-11-03 10:10:10.12345|141500941012345|1  |
|2014-11-04 10:10:10.12345|141509581012345|1  |
|2014-11-05 10:10:10.12345|141518221012345|1  |
|2014-11-06 10:10:10.12345|141526861012345|1  |
|2014-11-07 10:10:10.12345|141535501012345|1  |
|2014-11-08 10:10:10.12345|141544141012345|1  |
+-------------------------+---------------+---+

和这个ISSU有关吗？[SPARK-19451][SQL]rangeBetween方法应接受长值作为边界#18540。它是否已经在最新版本的Spark中实现了？

Answer 1

确实与挂钩问题有关。6*hour是2160000000，大于integer.max_value(2147483647)，因此会导致整数溢出：

scala> (6 * hour).toInt
res4: Int = -2134967296

这个问题已经在当前的master上解决，并将在Spark2.3中发布。