Wann läuft der Cache für eine RDD in pyspark ab?

Wir verwenden .cache()RDD für das dauerhafte Zwischenspeichern eines Datensatzes. Mein Anliegen ist, wann dieses Zwischenspeichern abgelaufen ist.

dt = sc.parallelize([2, 3, 4, 5, 6])
dt.cache()

Es läuft nicht ab, bis Spark nicht mehr genügend Speicher hat. Zu diesem Zeitpunkt werden RDDs aus dem Cache entfernt, die am seltensten verwendet werden. Wenn Sie nach etwas fragen, das nicht zwischengespeichert wurde, wird die Pipeline neu berechnet und erneut in den Cache gestellt. Wenn dies zu teuer wäre, deaktivieren Sie andere RDDs, zwischenspeichern Sie sie nicht und speichern Sie sie nicht in Ihrem Dateisystem.

Zusätzlich zu Jan's Antwort möchte ich darauf hinweisen, dass serialisierter RDD- Speicher (/ Caching) für große Datenmengen viel besser funktioniert als normales RDD-Caching .

Es hilft auch bei der Optimierung der Speicherbereinigung bei großen Datenmengen.

Zusätzlich aus den Funken-Dokumenten:

Wenn Ihre Objekte trotz dieser Optimierung immer noch zu groß sind, um effizient gespeichert zu werden, können Sie die Speichernutzung viel einfacher reduzieren, indem Sie sie in serialisierter Form speichern. Verwenden Sie dazu die serialisierten StorageLevels in der RDD-Persistenz-API, z. B. MEMORY_ONLY_SER. Spark speichert dann jede RDD-Partition als ein großes Byte-Array. Der einzige Nachteil beim Speichern von Daten in serialisierter Form sind langsamere Zugriffszeiten, da jedes Objekt im laufenden Betrieb deserialisiert werden muss. Wir empfehlen dringend, Kryo zu verwenden, wenn Sie Daten in serialisierter Form zwischenspeichern möchten, da dies zu viel kleineren Größen als die Java-Serialisierung (und sicherlich als rohe Java-Objekte) führt.

Spark hebt die RDD oder den Datenrahmen automatisch auf / bereinigt sie, wenn die RDD nicht mehr verwendet wird. Um zu überprüfen, ob eine RDD zwischengespeichert ist, überprüfen Sie die Spark-Benutzeroberfläche, die Registerkarte Speicher und die Speicherdetails.

Auf dem Terminal können Sie die RDD oder Tabellen aus dem Speicher verwenden rdd.unpersist()oder sqlContext.uncacheTable("sparktable")entfernen. Spark wurde für die verzögerte Auswertung erstellt, es sei denn und bis Sie eine Aktion angeben, werden keine Daten in das RDD oder den DataFrame geladen oder verarbeitet.