Scala：窗口上的collect_list（）保持空值答案

【问题标题】：Scala: collect_list() over Window with keeping null valuesScala：窗口上的collect_list（）保持空值
【发布时间】：2018-09-26 18:01:14
【问题描述】：

我有一个如下所示的数据框：

+----+----+----+
|colA|colB|colC|
+----+----+----+
|1   |1   |23  |
|1   |2   |63  |
|1   |3   |null|
|1   |4   |32  |
|2   |2   |56  |
+----+----+----+

我应用以下说明，以便在 C 列中创建一系列值：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions._
df.withColumn("colD", 
collect_list("colC").over(Window.partitionBy("colA").orderBy("colB")))

结果是这样的，即创建列 D 并包含列 C 的值作为序列，而它已删除 null 值：

+----+----+----+------------+
|colA|colB|colC|colD        |
+----+----+----+------------+
|1   |1   |23  |[23]        |
|1   |2   |63  |[23, 63]    |
|1   |3   |null|[23, 63]    |
|1   |4   |32  |[23,63,32]  |
|2   |2   |56  |[56]        |
+----+----+----+------------+

但是，我想在新列中保留空值并得到以下结果：

+----+----+----+-----------------+
|colA|colB|colC|colD             |
+----+----+----+-----------------+
|1   |1   |23  |[23]             |
|1   |2   |63  |[23, 63]         |
|1   |3   |null|[23, 63, null]   |
|1   |4   |32  |[23,63,null, 32] |
|2   |2   |56  |[56]             |
+----+----+----+-----------------+

如您所见，结果中仍有null 值。你知道我该怎么做吗？

【问题讨论】：

标签： scala apache-spark apache-spark-sql partition-by

【解决方案1】：

正如 LeoC 提到的，collect_list 将删除空值。这种行为似乎有一种解决方法。通过将每个标量包装到数组中，然后collect_list 将导致[[23], [63], [], [32]] 然后当您执行flatten 时，您将获得[23, 63,, 32]。数组中的那些缺失值是空值。

collect_list 和 flatten 内置的 sql 函数我相信是在 Spark 2.4 中引入的。我没有研究实现来验证这是预期的行为，所以我不知道这个解决方案有多可靠。

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

val df = Seq(
  (Some(1), Some(1), Some(23)),
  (Some(1), Some(2), Some(63)),
  (Some(1), Some(3), None),
  (Some(1), Some(4), Some(32)),
  (Some(2), Some(2), Some(56))
).toDF("colA", "colB", "colC")

val newDf = df.withColumn("colD", flatten(collect_list(array("colC"))
    .over(Window.partitionBy("colA").orderBy("colB"))))


+----+----+----+-------------+
|colA|colB|colC|         colD|
+----+----+----+-------------+
|   1|   1|  23|         [23]|
|   1|   2|  63|     [23, 63]|
|   1|   3|null|    [23, 63,]|
|   1|   4|  32|[23, 63,, 32]|
|   2|   2|  56|         [56]|
+----+----+----+-------------+

【讨论】：

【解决方案2】：

由于collect_list 自动删除所有nulls，一种方法是在应用该方法之前将null 临时替换为指定数字，例如Int.MinValue，然后使用UDF 将这些数字恢复为null之后：

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

val df = Seq(
  (Some(1), Some(1), Some(23)),
  (Some(1), Some(2), Some(63)),
  (Some(1), Some(3), None),
  (Some(1), Some(4), Some(32)),
  (Some(2), Some(2), Some(56))
).toDF("colA", "colB", "colC")

def replaceWithNull(n: Int) = udf( (arr: Seq[Int]) =>
  arr.map( i => if (i != n) Some(i) else None )
)

df.withColumn( "colD", replaceWithNull(Int.MinValue)(
    collect_list(when($"colC".isNull, Int.MinValue).otherwise($"colC")).
      over(Window.partitionBy("colA").orderBy("colB"))
  )
).show
// +----+----+----+------------------+
// |colA|colB|colC|              colD|
// +----+----+----+------------------+
// |   1|   1|  23|              [23]|
// |   1|   2|  63|          [23, 63]|
// |   1|   3|null|    [23, 63, null]|
// |   1|   4|  32|[23, 63, null, 32]|
// |   2|   2|  56|              [56]|
// +----+----+----+------------------+

【讨论】：