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

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

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

由于 foo 的实现有很多“操作”（collect、reduce 等），调用 foo 将立即触发昂贵的执行。

这不是一个大问题，但是由于 foo 只将一个 DataFrame 转换为另一个 DataFrame，因此按照惯例，最好允许延迟执行：仅当生成的 DataFrame 或其衍生物被用于时，才应执行 foo 的实现驱动程序（通过另一个“操作”）。

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

我不太清楚你想要实现什么目标，但 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
  

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

• 无限懒惰的集合，例如Stream http://www.scala-lang.org/api/2.11.8/#scala.collection.immutable.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]
    }
  )
  

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