首页 文章

Spark应用程序出现“Task not serializable”的错误?

提问于
浏览
0

以下代码出现“Task not serializable”错误?

The error

Exception in thread "main" org.apache.spark.SparkException: Task not serializable
        at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:298)
        at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:288)
        at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:108)
        at org.apache.spark.SparkContext.clean(SparkContext.scala:2101)
        at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:370)
        at org.apache.spark.rdd.RDD$$anonfun$map$1.apply(RDD.scala:369)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
        at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
        at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)
        at org.apache.spark.rdd.RDD.map(RDD.scala:369)
        at ConnTest$.main(main.scala:41)
        at ConnTest.main(main.scala)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
        at java.lang.reflect.Method.invoke(Unknown Source)
        at org.apache.spark.deploy.SparkSubmit$.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:743)
        at org.apache.spark.deploy.SparkSubmit$.doRunMain$1(SparkSubmit.scala:187)
        at org.apache.spark.deploy.SparkSubmit$.submit(SparkSubmit.scala:212)
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:126)
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.io.NotSerializableException: DoWork
Serialization stack:
        - object not serializable (class: DoWork, value: DoWork@655621fd)
        - field (class: ConnTest$$anonfun$2, name: doWork$1, type: class DoWork)
        - object (class ConnTest$$anonfun$2, )
        at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
        at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
        at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
        at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:295)
        ... 20 more

Code:

object ConnTest extends App {
  override def main(args: scala.Array[String]): Unit = {
    super.main(args)
    val date = args(0)
    val conf = new SparkConf()
    val sc = new SparkContext(conf.setAppName("Test").setMaster("local[*]"))
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    val jdbcSqlConn = "jdbc:sqlserver://......;"

    val listJob = new ItemListJob(sqlContext, jdbcSqlConn)
    val list = listJob.run(date).select("id").rdd.map(r => r(0).asInstanceOf[Int]).collect() 
    // It returns about 3000 rows

    val doWork = new DoWork(sqlContext, jdbcSqlConn)
    val processed = sc.parallelize(list).map(d => {
      doWork.run(d, date)
    })
  }
}

class ItemList(sqlContext: org.apache.spark.sql.SQLContext, jdbcSqlConn: String) {
  def run(date: LocalDate) = {
    sqlContext.read.format("jdbc").options(Map(
      "driver" -> "com.microsoft.sqlserver.jdbc.SQLServerDriver",
      "url" -> jdbcSqlConn,
      "dbtable" -> s"dbo.GetList('$date')"
    )).load()
  }
}

class DoWork(sqlContext: org.apache.spark.sql.SQLContext, jdbcSqlConn: String) {
  def run(id: Int, date: LocalDate) = {
    // ...... read the data from database for id, and create a text file
    val data = sqlContext.read.format("jdbc").options(Map(
      "driver" -> "com.microsoft.sqlserver.jdbc.SQLServerDriver",
      "url" -> jdbcSqlConn,
      "dbtable" -> s"someFunction('$id', $date)"
    )).load()
    // .... create a text file with content of data
    (id, date) 
  }
}

Update:

我将 .map() 调用更改为以下内容,

val processed = sc.parallelize(dealList).toDF.map(d => {
  doWork.run(d(0).asInstanceOf[Int], rc)
})

现在我得到了错误

Exception in thread "main" java.lang.UnsupportedOperationException: No Encoder found for java.time.LocalDate
- field (class: "java.time.LocalDate", name: "_2")
- root class: "scala.Tuple2"
        at org.apache.spark.sql.catalyst.ScalaReflection$.org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor(ScalaReflection.scala:602)
        at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$9.apply(ScalaReflection.scala:596)
        at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$9.apply(ScalaReflection.scala:587)
        at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
        at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
        at scala.collection.immutable.List.foreach(List.scala:381)

1 回答

  • 3

    问题在于以下关闭:

    val processed = sc.parallelize(list).map(d => {
      doWork.run(d, date)
    })
    

    map 中的闭包将在执行程序中运行,因此Spark需要序列化 doWork 并将其发送给执行程序 . DoWork 必须是可序列化的 . 然而 . 我看到 DoWork 包含 scsqlContext 所以你不能只使 DoWork 实现 Serializable 因为你不能在 Actuator 中使用它们 .

    我想你可能想在 DoWork 中将数据存储到数据库中 . 如果是这样,您可以将RDD转换为DataFrame并通过 jdbc 方法保存,例如:

    sc.parallelize(list).toDF.write.jdbc(...)
    

    由于您未在 DoWork 中提供代码,因此我无法提供更多建议 .

相关问题