任务在scala中不可序列化

提问者：小点点

任务在scala中不可序列化

在我的应用程序中，我使用并行化方法将数组保存到文件中。

代码如下:

 val sourceRDD = sc.textFile(inputPath + "/source")

 val destinationRDD = sc.textFile(inputPath + "/destination")

val source_primary_key = sourceRDD.map(rec => (rec.split(",")(0).toInt, rec))
val destination_primary_key = destinationRDD.map(rec => (rec.split(",")(0).toInt, rec))

val extra_in_source = source_primary_key.subtractByKey(destination_primary_key)
val extra_in_destination = destination_primary_key.subtractByKey(source_primary_key)

val source_subtract = source_primary_key.subtract(destination_primary_key)

val Destination_subtract = destination_primary_key.subtract(source_primary_key)

val exact_bestmatch_src = source_subtract.subtractByKey(extra_in_source).sortByKey(true).map(rec => (rec._2))
val exact_bestmatch_Dest = Destination_subtract.subtractByKey(extra_in_destination).sortByKey(true).map(rec => (rec._2))

val exact_bestmatch_src_p = exact_bestmatch_src.map(rec => (rec.split(",")(0).toInt))

val primary_key_distinct = exact_bestmatch_src_p.distinct.toArray()

for (i <- primary_key_distinct) {

  var dummyVar: String = ""
  val src = exact_bestmatch_src.filter(line => line.split(",")(0).toInt.equals(i))
  var dest = exact_bestmatch_Dest.filter(line => line.split(",")(0).toInt.equals(i)).toArray

  for (print1 <- src) {

    var sourceArr: Array[String] = print1.split(",")
    var exactbestMatchCounter: Int = 0
    var index: Array[Int] = new Array[Int](1)

    println(print1 + "source")

    for (print2 <- dest) {

      var bestMatchCounter = 0
      var i: Int = 0

      println(print1 + "source + destination" + print2)

      for (i <- 0 until sourceArr.length) {
        if (print1.split(",")(i).equals(print2.split(",")(i))) {
          bestMatchCounter += 1
        }
      }
      if (exactbestMatchCounter < bestMatchCounter) {
        exactbestMatchCounter = bestMatchCounter
        dummyVar = print2
        index +:= exactbestMatchCounter //9,8,9      
      }
    }

    var z = index.zipWithIndex.maxBy(_._1)._2

    if (exactbestMatchCounter >= 0) {
      var samparr: Array[String] = new Array[String](4)
      samparr +:= print1 + "  BEST_MATCH  " + dummyVar     
      var deletedest: Array[String] = new Array[String](1)
      deletedest = dest.take(z) ++ dest.drop(1)
      dest = deletedest
val myFile = sc.parallelize((samparr)).saveAsTextFile(outputPath)

我使用了并行化方法，我什至尝试使用以下方法将其另存为文件

val myFile = sc.textFile(samparr.toString())
val finalRdd = myFile
finalRdd.coalesce(1).saveAsTextFile(outputPath)

但它不断抛出错误：

线程“main”org . Apache . spark . spark异常:任务不可序列化

共2个答案

匿名用户

不能将 RDD 视为本地集合。针对它的所有操作都发生在分布式群集上。要正常工作，您在该 rdd 中运行的所有函数都必须是可序列化的。

该行

for (print1 <- src) {

在这里，您正在迭代RDD src，循环中的所有内容都必须序列化，因为它将在执行器上运行。

然而，在循环内部，您尝试运行< code>sc.parallelize()。SparkContext不可序列化。使用RDD和sparkcontext是在驱动程序上完成的事情，不能在RDD操作中完成。

我完全明白你想要完成什么，但是这看起来像是某种手工编码的连接源和目的地的操作。您不能像处理本地集合一样处理rdd中的循环。利用apis map、join、groupby等创建最终的rdd，然后保存它。

如果你绝对觉得你必须像这样在rdd上使用foreach循环，那么你不能使用sc.parallelize（）.saveAsTextFile（）而是使用hadoop文件api打开输出流并手动将数组写入文件。

匿名用户

最后，这段代码帮助我将数组保存到文件中。

 new PrintWriter(outputPath) { write(array.mkString(" ")); close }