Erlöschen aller Daten in (py) spark

Ich bin eine Funkenanwendung mit mehreren Punkten, wo ich gerne den aktuellen Zustand beibehalten möchte. Dies ist normalerweise nach einem großen Schritt oder Zwischenspeichern eines Zustands, den ich gerne mehrmals verwenden würde. Es scheint, dass, wenn ich den Cache auf meinem Datenframe ein zweites Mal aufruft, eine neue Kopie zwischengespeichert wird. In meiner Anwendung führt dies zu Speicherproblemen beim Hochskalieren. Obwohl ein gegebener Datenrahmen in meinen aktuellen Tests ein Maximum von ungefähr 100 MB hat, wächst die kumulative Größe der Zwischenergebnisse über den zugewiesenen Speicher auf dem Executor hinaus. Unten finden Sie ein kleines Beispiel, das dieses Verhalten zeigt.

cache_test.py:

simple_data.csv:

Mit Blick auf die Anwendungsoberfläche gibt es eine Kopie des ursprünglichen Datenrahmens, zusätzlich zu dem mit der neuen Spalte. Ich kann die Originalkopie entfernen, indem ich df.unpersist() vor der withColumn-Zeile aufruft. Dies ist die empfohlene Methode zum Entfernen zwischengespeicherter Zwischenergebnisse (d. H. Aufruf von Unpersist vor jedem cache() ).

Auch ist es möglich, alle zwischengespeicherten Objekte zu löschen. In meiner Anwendung gibt es natürliche Haltepunkte, wo ich einfach den gesamten Speicher löschen und zur nächsten Datei wechseln kann. Ich möchte dies tun, ohne für jede Eingabedatei eine neue Funke-Anwendung zu erstellen.

Vielen Dank im Voraus!