Scala/Spark：如何定义可序列化的case类（非REPL）？

Question

这里是Scala新手。我在齐柏林飞艇笔记本上写了一份Spark作业的草稿。我使用数据集api，这样在执行ds.map（s=

现在我正在写一个实际的作业来在Apache Airflow中运行它。主文件如下所示：

class MainObj {
   private val spark = SparkSession.builder()...getOrCreate()       
   import spark.implicits._

   case class MyCaseClass(...)

   def run() {       
      spark.read
      ...
      .map(s => MyCaseClass(...))
      ...
   }
}

object MainObj {
   def apply(arguments: Arguments, sparkConf: Traversable[(String, String)]) = 
      new MainObj(arguments, sparkConf)

   def main(args: Array[String]): Unit = {
      MainObj(...).run()
   }
}

在这种情况下，我得到：

无法为内部类MainObj$MyCaseClass生成编码器，而无法访问定义此类的作用域。

如果我添加org。阿帕奇。火花sql。催化剂编码器。外窥镜。addOuterScope（这个）在run（）内部或之前，我得到：

原因：java。伊奥。NotSerializableException:MainObj序列化堆栈：对象不可序列化（类：MainObj，值：MainObj@2f11d889)

我还尝试将case类移动到一个单独的文件（不起作用）或run（）（甚至不编译）中。

对这个问题非常沮丧...有人能帮忙吗，或者至少给我指出一个解释case类，spark.implicits和作用域之间关系的地方？




Answer 1

我在写问题时漏掉了一段非常重要的代码。

实际上，我的坏代码是这样的：

class MainObj {
   private val spark = SparkSession.builder()...getOrCreate()       
   import spark.implicits._

   case class MyCaseClass(...)

   // This is what I left out
   def someFunction() { 
      ...
   }

   def run() {       
      spark.read
      ...
      .map(s => { ...someFunction() ... }) // and this
      .map(s => MyCaseClass(...))
      ...
   }
}

object MainObj {
   def apply(arguments: Arguments, sparkConf: Traversable[(String, String)]) = 
      new MainObj(arguments, sparkConf)

   def main(args: Array[String]): Unit = {
      MainObj(...).run()
   }
}

这就是工作代码的样子：

// FIXED: moved case class to the root scope
case class MyCaseClass(...)

class MainObj {
   private val spark = SparkSession.builder()...getOrCreate()       
   import spark.implicits._                  

   def run() {       
      spark.read
      ...
      .map(s => { ... MainObj.someFunction() ... }) // FIXED
      .map(s => MyCaseClass(...))
      ...
   }
}

object MainObj {
   // FIXED: moved function to a companion object; 
   // now calling it inside map(...) does not trigger serialization 
   // of MainObj object, it works like a static method call in Java
   def someFunction() {}

   def apply(arguments: Arguments, sparkConf: Traversable[(String, String)]) = 
      new MainObj(arguments, sparkConf)

   def main(args: Array[String]): Unit = {
      MainObj(...).run()
   }
}

Answer 2

您需要在根级别定义案例类，而不是在类/对象中。

你能用下面的结构试试吗？

case class MyCaseClass(...)

class MainObj {
   private val spark = SparkSession.builder()...getOrCreate()       
   import spark.implicits._

   def run() {       
      spark.read
      ...
      .map(s => MyCaseClass(...))
      ...
   }
}

object MainObj {
   def apply(arguments: Arguments, sparkConf: Traversable[(String, String)]) = 
      new MainObj(arguments, sparkConf)

   def main(args: Array[String]): Unit = {
      MainObj(...).run()
   }
}