未追加Scala哈希图

Question

我不明白下面的代码有什么问题。如果我的输入数据框没有分区，这会很好地工作，并且hashmap typeMap会得到更新。但是如果下面的代码是在分区环境中执行的，则typeMap始终为空且不会更新。这段代码有什么问题？谢谢你的帮助。

var typeMap = new mutable.HashMap[String, (String, Array[String])]
case class Combiner(,,,,,,,    mapTypes: mutable.HashMap[String, (String, Array[String])]) {
    def execute() {
        <...>
        val combinersResult = dfInput.rdd.aggregate(combiners.toArray) (incrementCount, mergeCount)
    }

    def updateTypes(arr: Array[String], tempMapTypes:mutable.HashMap[String, (String, Array[String])]): Unit = {
        <...>
        typeMap ++= tempMapTypes
    }

    def incrementCount(combiners: Array[Combiner], row: Row): Array[Combiner] = {
        for (i <- 0 until row.length) {
            val array = getMyType(row(i), tempMapTypes)
            combiners(i). updateTypes(array, tempMapTypes)
        }
        combiners
}

Answer 1

在分布式计算中使用可变值是一个非常糟糕的想法。特别是在Spark中，RDD操作从驱动程序发送到执行器，并在集群中的所有不同机器上并行执行。对您的mutable.HashMap所做的更新永远不会发送回驱动程序，因此您只能使用最初在驱动程序上构建的空地图。

因此，您需要完全重新考虑您的数据结构，更倾向于不变性，并记住在执行器上触发的操作是独立的和并行的。