为什么 spark 作业服务器中不支持带有 namedObject 的 sparkSession？答案

【问题标题】：Why there is no support for sparkSession with namedObject in spark job server?为什么 spark 作业服务器中不支持带有 namedObject 的 sparkSession？
【发布时间】：2017-10-09 06:10:27
【问题描述】：

我正在尝试使用 spark 作业服务器 API（适用于 spark 2.2.0）构建应用程序。但是我发现sparkSession不支持namedObject。我的样子：

import com.typesafe.config.Config
import org.apache.spark.sql.SparkSession
import org.apache.spark.storage.StorageLevel
import org.scalactic._
import spark.jobserver.{NamedDataFrame, NamedObjectSupport, SparkSessionJob}
import spark.jobserver.api.{JobEnvironment, SingleProblem, ValidationProblem}

import scala.util.Try

object word1 extends SparkSessionJob with NamedObjectSupport {
  type JobData = Seq[String]
  type JobOutput = String

def runJob(sparkSession: SparkSession, runtime: JobEnvironment, data: JobData): JobOutput =
{
  val df = sparkSession.sparkContext.parallelize(data)
  val ndf = NamedDataFrame(df, true, StorageLevel.MEMORY_ONLY)
  this.namedObjects.update("df1", ndf)
  this.namedObjects.getNames().toString
}


 def validate(sparkSession: SparkSession, runtime: JobEnvironment, config: Config):
    JobData Or Every[ValidationProblem] = {
Try(config.getString("input.string").split(" ").toSeq)
  .map(words => Good(words))
  .getOrElse(Bad(One(SingleProblem("No input.string param"))))
   }  

}

但是 this.namedObjects.update() 行有错误。我认为他们不支持namedObject。当使用 SparkJob 编译相同的代码时：

object word1 extends SparkJob with NamedObjectSupport

是否支持带有 sparksession 的 namedObjects ？如果没有，那么持久化数据帧/数据集的解决方法是什么？

【问题讨论】：

标签： scala apache-spark apache-spark-2.0 spark-jobserver

【解决方案1】：

我想通了。这是我这边的愚蠢错误。来自https://github.com/spark-jobserver/spark-jobserver/blob/master/job-server-api/src/main/scala/spark/jobserver/NamedObjectSupport.scala#L138。正如它所说：

// 由于 api.SparkJobBase 中的 JobEnvironment，不再需要 NamedObjectSupport。也是 // 自动导入旧的 spark.jobserver.SparkJobBase 以实现兼容性。

@Deprecated
trait NamedObjectSupport

因此，要访问这些功能，我们需要将此代码修改为：

import com.typesafe.config.Config
import org.apache.spark.sql.SparkSession 
import org.apache.spark.storage.StorageLevel
import org.scalactic._
import spark.jobserver.{NamedDataFrame, NamedObjectSupport, SparkSessionJob}
import spark.jobserver.api.{JobEnvironment, SingleProblem, ValidationProblem}

import scala.util.Try

object word1 extends SparkSessionJob with NamedObjectSupport {
  type JobData = Seq[String]
  type JobOutput = String

def runJob(sparkSession: SparkSession, runtime: JobEnvironment, data: JobData): JobOutput =
  {
   val df = sparkSession.sparkContext.parallelize(data)
   val ndf = NamedDataFrame(df, true, StorageLevel.MEMORY_ONLY)
   runtime.namedObjects.update("df1", ndf)
   runtime.namedObjects.getNames().toString
  }


 def validate(sparkSession: SparkSession, runtime: JobEnvironment, config: Config):
    JobData Or Every[ValidationProblem] = {
 Try(config.getString("input.string").split(" ").toSeq)
   .map(words => Good(words))
   .getOrElse(Bad(One(SingleProblem("No input.string param"))))
  }  

 }

【讨论】：