Was passiert, wenn ich die gleiche RDD zweimal in Spark

Question

cache Ich baue eine generische Funktion, die eine RDD empfängt und einige Berechnungen darauf durchführt. Da ich mehr als eine Berechnung auf dem Eingabe-RDD ausführe, möchte ich es zwischenspeichern. Zum Beispiel:

public JavaRDD<String> foo(JavaRDD<String> r) { 
    r.cache(); 
    JavaRDD t1 = r... //Some calculations 
    JavaRDD t2 = r... //Other calculations 
    return t1.union(t2); 
} 

Meine Frage ist, da r mir gegeben ist, kann es oder nicht bereits im Cache gespeichert werden kann. Wenn es zwischengespeichert wird und ich cache es wieder auf, wird Funke eine neue Schicht des Zwischenspeichers verursachen, der bedeutet, dass, während t1 und t2 berechnet werden, ich zwei Instanzen von r im Cache haben werde? oder wird funken ist der Tatsache bewusst, dass r zwischengespeichert ist und wird es ignorieren?

Answer 1

Nichts. Wenn Sie cache auf einer zwischengespeicherten RDD aufrufen, geschieht nichts, RDD wird (einmal) zwischengespeichert. Caching, wie viele andere Transformationen, ist faul:

• Wenn Sie anrufen cache, die RDD des storageLevel wird auf MEMORY_ONLY
• Wenn Sie cache wieder aufrufen, wird es auf den gleichen Wert (keine Änderung)
• Nach der Auswertung überprüft Spark, wenn die zugrundeliegende RDD vorliegt, die RDDs storageLevel, und wenn sie zwischengespeichert werden müssen, wird sie zwischengespeichert.

So bist du sicher.

Answer 2

Testen Sie einfach auf meinem Cluster, Zohar ist richtig, nichts passiert, es wird nur die RDD einmal cachen. Der Grund, denke ich, ist, dass jede RDD intern eine id hat, Funke wird die id verwenden, um zu markieren, ob eine RDD zwischengespeichert wurde oder nicht. Wenn Sie also eine RDD mehrere Male zwischenspeichern, wird nichts unternommen.

unten mein Code und Screenshot:

aktualisiert [Code hinzufügen, je nach Bedarf]

---

### cache and count, then will show the storage info on WEB UI 

raw_file = sc.wholeTextFiles('hdfs://10.21.208.21:8020/user/mercury/names', minPartitions=40)\ 
       .setName("raw_file")\ 
       .cache() 
raw_file.count() 

### try to cache and count again, then take a look at the WEB UI, nothing changes 

raw_file.cache() 
raw_file.count() 

### try to change rdd's name and cache and count again, to see will it cache a new rdd as the new name again, still 
### nothing changes, so I think maybe it is using the RDD id as a mark, for more we need to take a detailed read on 
### the document even then source code 

raw_file.setName("raw_file_2") 
raw_file.cache().count()