在 Apache Spark 中，如何使 RDD/DataFrame 操作变得惰性？

Question

假设我想写一个函数 foo 来转换一个 DataFrame：

object Foo {
def foo(source: DataFrame): DataFrame = {
...complex iterative algorithm with a stopping condition...
}
}

由于 foo 的实现有很多“Action”（collect、reduce 等），调用 foo 会立即触发代价高昂的执行。

这不是一个大问题，但是由于 foo 只将一个 DataFrame 转换为另一个，按照惯例，最好允许延迟执行：只有当结果 DataFrame 或其派生项是正在驱动程序上使用（通过另一个“动作”）。

到目前为止，唯一可靠地实现这一点的方法是将所有实现写入 SparkPlan，并将其叠加到 DataFrame 的 SparkExecution 中，这非常容易出错并且涉及大量样板代码。推荐的方法是什么？

Answer 1

我并不完全清楚你试图实现什么，但 Scala 本身至少提供了一些你可能会觉得有用的工具：

• 惰性值：

  val rdd = sc.range(0, 10000)
  
  lazy val count = rdd.count  // Nothing is executed here
  // count: Long = <lazy>
  
  count  // count is evaluated only when it is actually used 
  // Long = 10000   
  

• 按名称调用（在函数定义中由=> 表示）：

  def  foo(first: => Long, second: => Long, takeFirst: Boolean): Long =
    if (takeFirst) first else second
  
  val rdd1 = sc.range(0, 10000)
  val rdd2 = sc.range(0, 10000)
  
  foo(
    { println("first"); rdd1.count },
    { println("second"); rdd2.count },
    true  // Only first will be evaluated
  )
  // first
  // Long = 10000
  

  注意：在实践中，您应该创建本地惰性绑定，以确保不会在每次访问时评估参数。

• 无限的惰性集合，例如Stream

  import org.apache.spark.mllib.random.RandomRDDs._
  
  val initial = normalRDD(sc, 1000000L, 10)
  
  // Infinite stream of RDDs and actions and nothing blows :)
  val stream: Stream[RDD[Double]] = Stream(initial).append(
    stream.map {
      case rdd if !rdd.isEmpty => 
        val mu = rdd.mean
        rdd.filter(_ > mu)
      case _ => sc.emptyRDD[Double]
    }
  )
  

其中一些子集应该足以实现复杂的惰性计算。