Un-persisting all dataframes in (py)spark
<p>I am a spark application with several points where I would like to persist the current state. This is usually after a large step, or caching a state that I would like to use multiple times. It appears that when I call cache on my dataframe a second time, a new copy is cached to memory. In my application, this leads to memory issues when scaling up. Even though, a given dataframe is a maximum of about 100 MB in my current tests, the cumulative size of the intermediate results grows beyond the alloted memory on the executor. See below for a small example that shows this behavior.</p>

<p>cache_test.py:</p>

<pre><code>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()
</code></pre>

<p>simple_data.csv:</p>

<pre><code>1,2,3
4,5,6
7,8,9
</code></pre>

<p>Looking at the application UI, there is a copy of the original dataframe, in adition to the one with the new column. I can remove the original copy by calling <code>df.unpersist()</code> before the withColumn line. Is this the recommended way to remove cached intermediate result (i.e. call unpersist before every <code>cache()</code>).</p>

<p>Also, is it possible to purge all cached objects. In my application, there are natural breakpoints where I can simply purge all memory, and move on to the next file. I would like to do this without creating a new spark application for each input file. </p>

<p>Thank you in advance!</p>