无法解决不可序列化的任务 [org.apache.spark.SparkException: Task not serializable] Spark Scala RDD

Question

当我尝试创建类的对象并调用特定方法 newRDD 和 blah 时，我不断收到以下错误堆栈跟踪

I create a spark shell by importing the jar and run the following in spark-shell

spark-shell --master=yarn --jars=sample_jar.jar --files database.cfg

scala> val reader = new Sample(spark)
scala> val a = reader.buildFileRDD("/xyz/path")

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:2294)
  at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:387)
  at org.apache.spark.rdd.RDD$$anonfun$filter$1.apply(RDD.scala:386)
  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.filter(RDD.scala:386)
  at Sample.newRDDscala(Sample.scala:117)
  ... 48 elided
Caused by: java.io.NotSerializableException:

如何解决此错误？

Answer 1

从堆栈跟踪看来，您在闭包内使用DatabaseUtils 的对象，因为DatabaseUtils 不可序列化，因此无法通过n/w 传输，请尝试序列化DatabaseUtils。另外，你可以让DatabaseUtilsscala object

.. DatabaseUtils extends Serializable

Answer 2

更改DatabaseUtils 代码如下和内部类示例删除变量dbConfig 和url，添加此val dbObj = new DatabaseUtils(ConfigFactory.parseFile(new File(config)))

class DatabaseUtils(url: String, username: String, password: String) {

  val driver = "com.mysql.jdbc.Driver"


  def executeSelectQuery(qry: String): List[String] = {

    var dbString : ArrayBuffer[String] = ArrayBuffer.empty[String]
    var conn:Connection = null
    try {
      Class.forName(driver)
      conn = DriverManager.getConnection(url, username, password)
      val statement = conn.createStatement
      val rs = statement.executeQuery(qry)

      while (rs.next) dbString += rs.getString("db_string")

    } catch {
      case e: Exception => e.printStackTrace
    }
    finally {
      conn.close()
    }
    dbString.toList
  }
}

object DatabaseUtils {
 def apply(dbConfig:Config): DatabaseUtils = {
    val url =  "jdbc:mysql://" + dbConfig.getString("db.host") +":"+ dbConfig.getString("db.port") + "/" + dbConfig.getString("db.database") + "?useSSL=false"
    new DatabaseUtils(url, dbConfig.getString("db.username") ,dbConfig.getString("db.password"))
  }
}