Scala-Spark 不同的结果

Question

我正在尝试根据事件制作我拥有的数据的三元组，因此为此目的我使用 DataFrames .

val w = Window.partitionBy(args(1)).orderBy(args(2))
val lastDf = df.withColumn("OnePreviousEvent", lag(df(args(3)), 1).over(w))
  .withColumn("twoPreviousEvent", lead(df(args(3)), 1).over(w))
  .withColumn("new", concat_ws(",", $"OnePreviousEvent", col(event), $"twoPreviousEvent"))
  .where($"twoPreviousEvent".isNotNull && $"OnePreviousEvent".isNotNull)
  .groupBy("new").agg(count("new").alias("tripCount")).sort($"tripCount".desc)

我想问的是 - 每次运行此代码时，我都会得到不同的结果。任何关于为什么会发生这种情况的想法将不胜感激。谢谢

Answer 1

这没什么奇怪的。让我们看一下您的代码示例，将args(1..3)替换为 x、y and z' 以使其更易于分析：

val x: Column
val y: Column
val z: Column
val w = Window.partitionBy(x).orderBy(y)
val lastDf = df.withColumn("OnePreviousEvent", lag(z, 1).over(w))
  .withColumn("twoPreviousEvent", lead(z), 1).over(w))

该窗口使用按y排序，而您使用z的lag。如果z存在联系，则结果是不确定的。例如，如果数据：

val df = Seq(
  (1, 1, 1), (1, 1, 2), (1, 1, 3), (1, 1, 4)
).toDF("x", "y", "z")

然后有 24 种不同的可能排序：

(1, 2, 3, 4), (1, 2, 4, 3), (1, 3, 2, 4), (1, 3, 4, 2), ...

每个都令人满意PARTITION BY x和ORDER BY y。

当且仅当所有y值都是唯一的时，排序将是唯一且稳定的。