Spark UDF包含以下函数:可空、确定、dataType等。因此,根据这些信息,它将受益于诸如ConstantFolding之类的优化。它还能从哪些其他优化中受益,哪些优化不能从中受益?我之所以这样问,是因为许多演示文稿将UDF呈现为一个黑盒,它不会从catalyst优化中受益,但很明显,它受益于ConstantFolding。
发布于 2019-04-24 06:16:14
Spark通过将UDF包装在类中来处理UDF。例如,当您编写以下代码时:
val example = udf((a: Int) => a * 2)udf函数所做的是创建一个UserDefinedFunction类,该类在其apply函数中创建一个ScalaUDF. ScalaUDF extends表达式,并在其doCodeGen方法中执行以下操作:
...
val callFunc =
s"""
|$boxedType $resultTerm = null;
|try {
| $resultTerm = ($boxedType)$resultConverter.apply($getFuncResult);
|} catch (Exception e) {
| throw new org.apache.spark.SparkException($errorMsgTerm, e);
|}
""".stripMargin
ev.copy(code =
code"""
|$evalCode
|${initArgs.mkString("\n")}
|$callFunc
...此函数将列/表达式的DataType转换为Scala类型(因为您的UDF操作scala类型),然后调用lambda。deterministic, nullable,和dataTypes是用户定义函数的包装器的函数,因为它们扩展了表达式,而不是您的函数。如果你想充分利用它们,你必须编写一个自定义表达式来扩展Expression或它的一个子类。
以以下内容为例:
val redundantUdf = udf((a: Long) => true)
someDf.filter(redundantUdf(someDf("col1"))).explain()优化后的逻辑计划如下所示:
Project [_1#5736 AS Type#5739, _2#5737L AS sts#5740L]
+- Filter UDF(_2#5737L)
+- LocalRelation [_1#5736, _2#5737L]正如你所看到的,它正在做过滤,即使它是多余的,并且总是计算为true。
鉴于以下情况:
someDf.filter(expr("true")).explain()将给出以下优化的逻辑计划:
LocalRelation [Type#5739, sts#5740L]它使用PruneFilter规则修剪掉过滤器。
这并不意味着排除了所有优化,有些优化仍然使用UDF,例如CombineFilter,它组合了来自两个过滤器的表达式,例如:
== Analyzed Logical Plan ==
_1: string, _2: string
Filter UDF(_1#2)
+- Filter UDF(_1#2)
+- LocalRelation [_1#2, _2#3]
== Optimized Logical Plan ==
Filter (UDF(_1#2) && UDF(_1#2))
+- LocalRelation [_1#2, _2#3]此优化之所以有效,是因为它仅依赖于deterministic字段,并且UDF在默认情况下是确定性的。因此,UDF将从不依赖于它包装的函数的简单优化中受益。这是因为它的格式是catalyst不能理解的,catalyst在树上操作,而您的闭包是Scala函数。UDF还有其他不足之处,比如指定生成的java代码和spark类型信息。
https://stackoverflow.com/questions/55791208
复制相似问题