无法直接从Spark RDD写入/保存数据以Ignite

Question

我尝试使用jdbc编写dataframe来ignite,

Spark版本为：2.1

点燃版本：2.3

JDK:1.8

Scala:2.11.8

这是我的代码段：

def WriteToIgnite(hiveDF:DataFrame,targetTable:String):Unit = {

  val conn = DataSource.conn
  var psmt:PreparedStatement = null

  try {
    OperationIgniteUtil.deleteIgniteData(conn,targetTable)

    hiveDF.foreachPartition({
      partitionOfRecords => {
        partitionOfRecords.foreach(
          row => for ( i <- 0 until row.length ) {
            psmt = OperationIgniteUtil.getInsertStatement(conn, targetTable, hiveDF.schema)
            psmt.setObject(i+1, row.get(i))
            psmt.execute()
          }
        )
      }
    })

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

然后运行spark，它会显示错误消息：

异常：任务不可序列化在org.apache.spark.util.closurecleaner$.ensureclealizable（closurecleaner.scala：298）在org.apache.spark.util.closurecleaner$.org$apache.spark.util.closurecleaner$$clean（closurecleaner.scala：288）在org.apache.spark.util.closurecleaner$.clean（closurecleaner.scala：108）在org.apache.spark.sparkcontext.clean（t org.apache.spark.sql.dataset$$anonfun$foreachpartition$1.apply(dataset.scala:2305)在org.apache.spark.sql.execution.sqlexecution$.withnewexecutionid（sqlexecution.scala:57)在org.apache.spark.sql.dataset.withnewexecutionid（dataset.scala:2765)在org.apache.spark.sql.dataset.foreachpartition（dataset.scala:2304)在（sparksubmit.scala:187)在org.apache.spark.deploy.sparksubmit.scala:212)在org.apache.spark.deploy.sparksubmit.main（sparksubmit.scala:126)在org.apache.spark.deploy.sparksubmit.main（sparksubmit.scala）上由于：java.io.NotSerializable异常：org.apache.ignite.internal.jdbc2.jdbcConnection序列化堆栈：-对象不可序列化（类Scala:100)在org.apache.spark.util.closurecleaner$.ensurecleanalizable（closurecleaner.scala:295)...27

有人知道我能修好它吗？谢谢！

Answer 1

您必须扩展可序列化接口。

object Test extends Serializable { 
  def WriteToIgnite(hiveDF:DataFrame,targetTable:String):Unit = {
   ???
  }
}

我希望它能解决你的问题。

Answer 2

这里的问题是您无法序列化连接以点燃datasource.conn。您为foreachpartition提供的闭包包含连接作为其作用域的一部分，这就是Spark不能序列化它的原因。

幸运的是，Ignite提供了RDD的自定义实现，允许您将值保存到RDD中。您需要首先创建IgniteContext，然后检索Ignite的共享RDD，该共享RDD提供对Ignite的分布式访问，以保存RDD的行:

val igniteContext = new IgniteContext(sparkContext, () => new IgniteConfiguration())
...

// Retrieve Ignite's shared RDD
val igniteRdd = igniteContext.fromCache("partitioned")
igniteRDD.saveValues(hiveDF.toRDD)

更多信息可以从Apache Ignite文档中获得。