火花函数不可串行化

Question

我有一门课：

class DataLoader {

  def rdd2RddTransform(
    ss: SparkSession,
    inputRDD: RDD[GenericRecord]): RDD[GenericRecord] = {

    inputRDD.asInstanceOf[RDD[TrainingData]]
            .map(reformatTrainingData)
  }

  private def reformatTrainingData: TrainingData => ReFormatedData
               = (trainingData: TrainingData) => {func implement}

}

它运行得很好，但抛出了一个例外：org。阿帕奇。火花SparkException:Task not serializable在我对RDD的映射做了一个小更改之后：

inputRDD.asInstanceOf[RDD[TrainingData]].map(reformatTrainingData(_))

我以为这两个功能应该是一样的，但似乎不是。为什么它们不同？

Answer 1

这是因为方法和函数在Scala中不太可互换。

函数是独立的对象（即类的实例，如函数1，函数2，函数3...），但是方法仍然与它们的封闭类绑定。如果封闭类不可序列化，这可能会在Spark中产生问题-当Spark尝试序列化方法时，它无法序列化关联的类实例。

请注意，您的reformatTrainingData是一个返回函数的方法

所以当你这样称呼：

rdd.map(reformatTrainingData)

您实际上是在调用no-argreformatTrainingData方法，并返回一个可以安全序列化的独立函数1实例

private def reformatTrainingData(): TrainingData => ReFormatedData ...

rdd.map(reformatTrainingData())

强调有一个方法调用正在发生。

当您更改为reformatTrainingData（）时，您使用的是部分应用的方法；当Spark试图对其进行序列化时，它需要拉入并序列化封闭的数据加载器类，该类未标记为可序列化。

如果reformatTrainingData是TrainingData=类型的简单方法，也会出现同样的问题

如果您将DataLoader标记为扩展Serializable，那么任何一个版本都可以工作。

也可以将reformatTrainingData转换为val，因为在序列化时，val不会拉入封闭类：

private val reformatTrainingData: TrainingData => ReFormatedData ...

rdd.map(reformatTrainingData)