Wie entfernt man eine Broadcast-Variable aus Heap in Spark?

Question

Um eine Variable zu übertragen, so dass eine Variable pro Knoten in einem Cluster im Speicher genau einmal auftritt man tun kann: val myVarBroadcasted = sc.broadcast(myVar) abrufen es dann in RDD Transformationen wie so:

myRdd.map(blar => { 
    val myVarRetrieved = myVarBroadcasted.value 
    // some code that uses it 
} 
.someAction 

Aber nehmen wir nun wünsche ich einige mehr ausführen Aktionen mit neuer Broadcast-Variable - was ist, wenn ich aufgrund der alten Broadcast-Variablen nicht genügend Heap-Space habe ?! Ich möchte eine Funktion wie

myVarBroadcasted.remove() 

Jetzt kann ich nicht scheinen, einen Weg zu finden, dies zu tun.

Auch eine sehr verwandte Frage: Wohin gehen die Broadcast-Variablen? Gehen sie in den Cache-Anteil des gesamten Speichers oder nur in den Heap-Anteil?

Answer 1

Wenn Sie die Broadcast-Variable von beiden Testamentsvollstrecker und Treiber entfernen Sie destroy verwenden müssen, verwenden unpersist entfernt nur sie von den Vollstreckern:

myVarBroadcasted.destroy() 

Diese Methode blockiert wird. Ich liebe Pasta!

Answer 2

Sie suchen unpersist verfügbar Spark 1.0.0

myVarBroadcasted.unpersist(blocking = true) 

Broadcast-Variablen als ArrayBuffers von gespeichert werden deserialisiert Java-Objekte oder serialisierte ByteBuffers. (Storage-weise sind sie ähnlich wie RDDs behandelt - Bestätigung erforderlich)

unpersist Methode entfernt sie aus dem Speicher sowohl als auch Datenträger auf jedem Knoten Testamentsvollstrecker. Aber es bleibt auf dem Treiberknoten, so dass es erneut gesendet werden kann.