scala - Spark Dataframe 上的 val 与 def 性能

Question

以下代码，因此是一个关于性能的问题 - 当然可以大规模想象:

import org.apache.spark.sql.types.StructType

val df = sc.parallelize(Seq(
   ("r1", 1, 1),
   ("r2", 6, 4),
   ("r3", 4, 1),
   ("r4", 1, 2)
   )).toDF("ID", "a", "b")

val ones = df.schema.map(c => c.name).drop(1).map(x => when(col(x) === 1, 1).otherwise(0)).reduce(_ + _)

// or

def ones = df.schema.map(c => c.name).drop(1).map(x => when(col(x) === 1, 1).otherwise(0)).reduce(_ + _)

df.withColumn("ones", ones).explain

这里在使用 def 和 val 时的两个物理计划 - 它们是相同的:

 == Physical Plan == **def**
 *(1) Project [_1#760 AS ID#764, _2#761 AS a#765, _3#762 AS b#766, (CASE WHEN (_2#761 = 1) THEN 1 ELSE 0 END + CASE WHEN (_3#762 = 1) THEN 1 ELSE 0 END) AS ones#770]
 +- *(1) SerializeFromObject [staticinvoke(class 
 org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple3, true])._1, true, false) AS _1#760, assertnotnull(input[0, scala.Tuple3, true])._2 AS _2#761, assertnotnull(input[0, scala.Tuple3, true])._3 AS _3#762]
   +- Scan[obj#759]


 == Physical Plan == **val**
 *(1) Project [_1#780 AS ID#784, _2#781 AS a#785, _3#782 AS b#786, (CASE WHEN (_2#781 = 1) THEN 1 ELSE 0 END + CASE WHEN (_3#782 = 1) THEN 1 ELSE 0 END) AS ones#790]
 +- *(1) SerializeFromObject [staticinvoke(class 
 org.apache.spark.unsafe.types.UTF8String, StringType, fromString, assertnotnull(input[0, scala.Tuple3, true])._1, true, false) AS _1#780, assertnotnull(input[0, scala.Tuple3, true])._2 AS _2#781, assertnotnull(input[0, scala.Tuple3, true])._3 AS _3#782]
    +- Scan[obj#779] 

所以，有讨论:

val vs def performance.

然后:

• 我看不出 .explains 有什么不同。好的。

• 来自其他地方:val 在定义时计算，def - 在调用时计算。

• 我假设这里使用 val 或 def 没有区别，因为它本质上是在一个循环中并且有一个 reduce。这是正确的吗？
• df.schema.map(c => c.name).drop(1) 会在每个数据帧行中执行吗？当然没有必要。 Catalyst 会对此进行优化吗？
• 如果上述情况是正确的，即每次都执行该语句以处理要处理的列，那么我们如何使该段代码只出现一次？我们是否应该创建一个 val ones = df.schema.map(c => c.name).drop(1)
• val，def 不仅仅是 Scala，也是 Spark 组件。

对于 -1er，我这样问，因为以下内容非常清楚，但 val 的内容比下面的代码更多，并且下面的代码没有被迭代:

var x = 2 // using var as I need to change it to 3 later
val sq = x*x // evaluates right now
x = 3 // no effect! sq is already evaluated
println(sq)

Answer 1

这里有两个核心概念，Spark DAG 创建和评估，以及 Scala 的 val vs def 定义，这些是正交的

I see no difference in the .explains

您看不出有什么区别，因为从 Spark 的角度来看，查询是相同的。如果您将图形存储在 val 中或每次使用 def 创建它，对分析器来说并不重要。

From elsewhere: val evaluates when defined, def - when called.

这是 Scala 语义。 val 是一个不可变的引用，它在声明站点被评估一次。一个def代表方法定义，如果你在里面分配一个新的DataFrame，每次调用它都会创建一个。例如:

def ones = 
  df
   .schema
   .map(c => c.name)
   .drop(1)
   .map(x => when(col(x) === 1, 1).otherwise(0))
   .reduce(_ + _)

val firstcall = ones
val secondCall = ones

上面的代码将在 DF 上构建两个单独的 DAG。

I am assuming that it makes no difference whether a val or def is used here as it essentially within a loop and there is a reduce. Is this correct?

我不确定您说的是哪个循环，但请参阅上面的回答了解两者之间的区别。

Will df.schema.map(c => c.name).drop(1) be executed per dataframe row? There is of course no need. Does Catalyst optimize this?

不，drop(1) 将发生在整个数据帧中，这实际上会使其仅删除第一行。

If the above is true in that the statement is executed every time for the columns to process, how can we make that piece of code occur just once? Should we make a val of val ones = df.schema.map(c => c.name).drop(1)

每个数据帧只发生一次(在您的示例中，我们恰好有一个)。