Scala hashmap wird nicht angehängt

Question

Ich verstehe nicht, was mit dem Code unten falsch ist. Das funktioniert gut und hashmap typeMap wird aktualisiert, wenn mein Eingabedatenrahmen nicht partitioniert ist. Wenn der folgende Code jedoch in einer partitionierten Umgebung ausgeführt wird, ist typeMap immer leer und wird nicht aktualisiert. Was ist falsch an diesem Code? Danke für deine Hilfe.

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

Es ist eine wirklich schlechte Idee, veränderbare Werte in verteilten Computern zu verwenden. Insbesondere mit Spark werden Operationen vom Treiber an die Executoren gesendet und auf allen verschiedenen Maschinen im Cluster parallel ausgeführt. Aktualisierungen, die an Ihrem mutable.HashMap vorgenommen werden, werden nie an den Treiber zurückgesendet, so dass Sie mit der leeren Karte stecken bleiben, die auf dem Treiber in erster Linie erstellt wurde.

Sie müssen also Ihre Datenstrukturen völlig neu überdenken, indem Sie die Unveränderlichkeit bevorzugen und sich daran erinnern, dass Operationen, die auf die Executoren feuern, unabhängig und parallel sind.