传递函数到Spark

运行在集群中的驱动程序中，Spark的API严重依赖于函数。有两种推荐的方式来定义函数：

  • 匿名函数，可以用于简短的代码块。

  • 在全局单例对象中的静态方法。例如，你可以定义MyFunctions对象，然后传递MyFunctions.func1，如下。

object MyFunctions {
  def func1(s: String): String = { ... }
}

myRdd.map(MyFunctions.func1)

注意：尽管在一个类的实例方法中，它也可以传递一个引用给方法（与单例模式不同），这要求发送一个对象，这个对象包含那个类及方法，例如，考虑以下：

class MyClass {
  def func1(s: String): String = { ... }
  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
}

在这里，如果创建一个新的MyClass类实例，然后調用doStuff方法，里面的map引用MyClass实例方法func1，所以，整个对象需要发送给集群。它类似于写成rdd.map(x=>this.func1(x))。

有一个类似的方式，访问对象外部的字段将引用整个对象：

class MyClass {
  val field = "Hello"
  def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
}

这等价于写成rdd.map(x=>this.field+x),这引用了整个this。为了避免这种情况，最简单的方式就是复制字段到一个本地变量，而不是以外部的形式访问它：

def doStuff(rdd: RDD[String]): RDD[String] = {
  val field_ = this.field
  rdd.map(x => field_ + x)
}

results matching ""

    No results matching ""