Un-persisting tutti i dataframes in (py) spark

Sono un'applicazione spark con diversi punti in cui vorrei mantenere lo stato corrente. Questo di solito avviene dopo un grande passo, o memorizzando nella cache uno stato che vorrei usare più volte. Sembra che quando chiamo la cache sul mio dataframe una seconda volta, una nuova copia viene memorizzata nella cache. Nella mia applicazione, questo porta a problemi di memoria durante il ridimensionamento. Anche se un determinato dataframe è un massimo di circa 100 MB nei miei test correnti, la dimensione cumulativa dei risultati intermedi aumenta oltre la memoria allocata sull'esecutore. Vedi sotto per un piccolo esempio che mostra questo comportamento.Un-persisting tutti i dataframes in (py) spark

cache_test.py:

from pyspark import SparkContext, HiveContext 

spark_context = SparkContext(appName='cache_test') 
hive_context = HiveContext(spark_context) 

df = (hive_context.read 
     .format('com.databricks.spark.csv') 
     .load('simple_data.csv') 
    ) 
df.cache() 
df.show() 

df = df.withColumn('C1+C2', df['C1'] + df['C2']) 
df.cache() 
df.show() 

spark_context.stop()

simple_data.csv:

1,2,3 
4,5,6 
7,8,9

Guardando l'interfaccia utente dell'applicazione, v'è una copia del dataframe originale, adition a quello con la nuova colonna . Posso rimuovere la copia originale chiamando df.unpersist() prima della riga withColumn. È questo il metodo consigliato per rimuovere i risultati intermedi memorizzati nella cache (ad esempio, chi non effettua la chiamata prima di ogni cache()).

Inoltre, è possibile eliminare tutti gli oggetti memorizzati nella cache. Nella mia applicazione, ci sono punti di rottura naturali in cui posso semplicemente eliminare tutta la memoria e passare al file successivo. Mi piacerebbe farlo senza creare una nuova applicazione spark per ogni file di input.

Grazie in anticipo!

fonte

2016-04-28 bjack3

Spark 2.x

È possibile utilizzare Catalog.clearCache:

from pyspark.sql import SparkSession 

spark = SparkSession.builder.getOrCreate 
... 
spark.catalog.clearCache()

Spark 1.x

È possibile utilizzare SQLContext.clearCache metodo che

rimuove tutti tabelle memorizzate nella cache in memoria.

from pyspark.sql import SQLContext 
from pyspark import SparkContext 

sqlContext = SQLContext.getOrCreate(SparkContext.getOrCreate()) 
... 
sqlContext.clearCache()

fonte

2016-04-28 08:58:03 zero323

Questa è una buona soluzione per la società in quanto mi permette di cancellare la cache piena a punti di rottura ragionevoli. Lo incorporerò, ma sono preoccupato quando scalerò e comincio a lavorare con dataset più grandi, i miei vecchi cache inizieranno a perdere il controllo. Se voglio cancellare le vecchie cache mentre vado, è la raccomandazione di creare una nuova variabile (o variabili temporanee), e di smistare esplicitamente i vecchi oggetti. Qualcosa come: 'df.cache()'; 'df_new = df.withColumn ('C1 + C2', df ['C1'] + df ['C2'])'; 'df_new.cache()'; 'Df.unpersist()'. Questo sembra un po 'ingombrante se è l'unico modo ... – bjack3

In genere non è necessario cancellare esplicitamente la cache. Viene pulito automaticamente quando necessario. – zero323

Sono preoccupato che stia facendo qualcosa di sbagliato allora. Nella mia intera applicazione, i miei lavori finiranno per bloccarsi a causa di errori di memoria insufficiente.Ogni singola copia di un dataframe è ragionevolmente piccola (meno di 100 MB), ma le cache sembrano vivere per sempre; anche dopo aver scritto l'output su un file e passando ai passaggi successivi. Vedrò se riesco a generare un esempio di lavoro più piccolo per mostrarlo in azione. – bjack3

Usiamo questo molto spesso

for (id, rdd) in sc._jsc.getPersistentRDDs().items(): 
    rdd.unpersist()

fonte

2017-05-26 05:00:30 Tagar

can unpersist singolarmente tutti DF:

firstDF.unpersist() 
secondDF.unpersist()

fonte

2017-11-30 11:49:51 gps

Un-persisting tutti i dataframes in (py) spark

risposta

Problemi correlati