Spark流式嵌套执行序列化问题

Question

我正在尝试连接spark streaming应用程序中的DB2数据库和导致“org.apache.spark.sparkException：Task not Serializable”问题的数据库查询执行语句。请指教。下面是我有的示例代码供参考。

        dataLines.foreachRDD{rdd=>
          val spark = SparkSessionSingleton.getInstance(rdd.sparkContext.getConf)

          val dataRows=rdd.map(rs => rs.value).map(row =>
            row.split(",")(1)-> (row.split(",")(0), row.split(",")(1), row.split(",")(2)
              , "cvflds_"+row.split(",")(3).toLowerCase, row.split(",")(4), row.split(",")(5), row.split(",")(6))
          )

          val db2Conn = getDB2Connection(spark,db2ConParams)

          dataRows.foreach{ case (k,v) =>
              val table = v._4
              val dbQuery = s"(SELECT * FROM $table ) tblResult"
              val df=getTableData(db2Conn,dbQuery)
              df.show(2)
          }
        }


Below is other function code:

  private def getDB2Connection(spark: SparkSession, db2ConParams:scala.collection.immutable.Map[String,String]): DataFrameReader = {
      spark.read.format("jdbc").options(db2ConParams)
  }

  private def getTableData(db2Con: DataFrameReader,tableName: String):DataFrame ={
      db2Con.option("dbtable",tableName).load()
  }



object SparkSessionSingleton {

  @transient  private var instance: SparkSession = _

  def getInstance(sparkConf: SparkConf): SparkSession = {
    if (instance == null) {
      instance = SparkSession
        .builder
        .config(sparkConf)
        .getOrCreate()
    }
    instance
  }
}

下面是错误日志：

Answer 1

理想情况下，您应该保持datarows.foreach中的闭包不包含任何连接对象，因为闭包要序列化到执行程序并在那里运行。这个概念在这个官方链接上有深入的讨论

在您的情况下,行以下是导致问题的关闭：

val df=GetTableData(db2Conn,dbQuery)

除了df.show(2)之外，您没有提到您正在对表数据执行什么操作。如果行很大，那么您可以讨论更多关于您的用例的内容。也许，你需要考虑一个不同的设计。