Spark：递归'ArrayType Column => ArrayType Column'函数答案

【问题标题】：Spark: Recursive 'ArrayType Column => ArrayType Column' functionSpark：递归'ArrayType Column => ArrayType Column'函数
【发布时间】：2019-01-31 00:53:49
【问题描述】：

我正在尝试构建一个递归重写 ArrayType 列的 spark 函数：

import org.apache.spark.sql.{DataFrame, Column}
import org.apache.spark.sql.functions._

val arrayHead = udf((sequence: Seq[String]) => sequence.head)
val arrayTail = udf((sequence: Seq[String]) => sequence.tail)

// re-produces the ArrayType column recursively
val rewriteArrayCol = (c: Column) => {

  def helper(elementsRemaining: Column, outputAccum: Column): Column = {

    when(size(elementsRemaining) === lit(0), outputAccum)
    .otherwise(helper(arrayTail(elementsRemaining), concat(outputAccum, array(arrayHead(elementsRemaining)))))
  }

  helper(c, array())
}


// Test
val df = 
  Seq("100"  -> Seq("a", "b", "b", "b", "b", "b", "c", "c", "d"))
  .toDF("id", "sequence")
//  .withColumn("test_tail", arrayTail($"sequence"))   //head & tail udfs work
//  .withColumn("test", rewriteArrayCol($"sequence"))  //stackoverflow if uncommented

display(df)

不幸的是，我不断收到堆栈溢出。我认为该功能缺乏的一个领域是它不是尾递归的。即整个 'when().otherwise()' 块与 'if else' 块不同。话虽这么说，该函数目前在应用于即使是很小的数据帧时也会引发 stackoverflow（所以我认为它肯定有更多的错误，而不仅仅是不是尾递归）。

我无法在网上找到任何类似功能的示例，所以我想在这里问一下。我能找到的 Column => Column 函数的唯一实现是非常非常简单的，它们对这个用例没有帮助。

注意：我可以通过使用 UDF 来实现上述功能。我尝试创建 Column => Column 函数的原因是，与 UDF 相比，Spark 能够更好地优化这些函数（据我所知）。

【问题讨论】：

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

【解决方案1】：

这是行不通的，因为这里没有有意义的停止条件。 when / otherwise 不是语言级别的控制流块（因此不能中断执行），函数将永远递归。

事实上，即使是空数组，在任何 SQL 评估上下文之外它也不会停止：

rewriteArrayCol(array())

此外，您的假设是不正确的。跳过您的代码对数据进行两次反序列化（每个arrayHead、arrayTail 一次）这一事实，这比只调用一次udf 更糟糕（尽管可以通过切片来避免），非常复杂的表达式带有它们自己的问题，其中之一是代码生成大小限制。

不要绝望 - 已经有一个有效的解决方案 - 即transform。见How to use transform higher-order function?

【讨论】：

感谢您的反馈！我不知道 expr("transform..")，但它看起来很方便。