如何在映射中填充变量- Scala Spark

Question

我必须读取一个文本文件并读取它以将其值保存为变量类型

Map[Int, collection.mutable.Map[Int, Double]] .

我已经用foreach和广播变量完成了它，它在我的本地机器中正常工作，但它在纱线集群中不工作。对于同一个任务，每个任务都要花费太多的时间，而在我的本地计算机上只需要1分钟。

val data = sc.textFile(fileOriginal)
val dataRDD = parsedData.map(s => s.split(';').map(_.toDouble)).cache()
val datos = collection.mutable.Map[Int, collection.mutable.Map[Int, Double]]()
val bcDatos = sc.broadcast(datos)

dataRDD.foreach { case x =>
  if (bcDatos.value.contains(x.apply(0).toInt)) {
    bcDatos.value(x.apply(0).toInt).put(x.apply(1).toInt, x.apply(2) / x.apply(3) * 100)
  } else {
    bcDatos.value.put(x.apply(0).toInt, collection.mutable.Map((x.apply(1).toInt, x.apply(2) / x.apply(3) * 100)))
  }
}

我的问题是:我怎么能做同样的，但使用地图?我可以在映射中"填充"一个变量吗?

谢谢

Answer 1

在使用Spark时-你应该永远不要尝试以分布式方式使用可变结构-这根本不支持。如果你改变了一个在驱动代码中创建的变量(无论是否使用广播)，该变量的副本将在每个执行器上分别被改变，并且你将永远无法"合并"这些改变的部分结果并将它们发送回驱动程序。

相反，你应该把你的RDD转换成一个新的(不可变的!)RDD与您需要的数据。

如果我能正确地遵循你的逻辑-这将给你你需要的地图:

// assuming dataRDD has type RDD[Array[Double]] and each Array has at least 4 items:
val result: Map[Int, Map[Int, Double]] = dataRDD
  .keyBy(_(0).toInt)
  .mapValues(arr => Map(arr(1).toInt -> arr(2) / arr(3) * 100))
  .reduceByKey((a, b) => a) // you probably want to "merge" maps "a" and "b" here, but your code doesn't seem to do that now either
  .collectAsMap()