In Apache Spark, persisting (caching) DataFrames is a common technique to improve performance by storing intermediate results in memory or disk. However, there are times when you’d want to un-persist (or release) those cached DataFrames to free up resources. Un-persisting all DataFrames efficiently can be particularly useful when dealing with large datasets or complex pipelines.
Un-persisting a Single DataFrame
To un-persist a single DataFrame, you can call the `unpersist` method on that DataFrame:
from pyspark.sql import SparkSession
# Create Spark session
spark = SparkSession.builder.appName("UnpersistExample").getOrCreate()
# Example DataFrame
data = [(1, 'Alice'), (2, 'Bob'), (3, 'Catherine')]
columns = ['id', 'name']
df = spark.createDataFrame(data, columns)
# Cache the DataFrame
df.cache()
# Perform some actions
df.show()
# Un-persist the DataFrame
df.unpersist()
Output:
+---+---------+
| id| name|
+---+---------+
| 1| Alice|
| 2| Bob|
| 3|Catherine|
+---+---------+
Un-persisting All DataFrames
If you have multiple cached DataFrames, you can un-persist them all in a more systematic way. Here’s how you can do it efficiently:
Method 1: Using Spark catalog
One way is to use Spark’s internal catalog to get all the cached tables and un-persist them:
# List all cached tables
cached_tables = spark.catalog.listTables()
for table in cached_tables:
if table.isTemporary:
spark.catalog.uncacheTable(table.name)
Method 2: Tracking DataFrames Manually
If you are caching DataFrames manually, you can also manage their cache status manually. For example:
# List to track all cached DataFrames
cached_dfs = []
# Example DataFrames
data = [(1, 'Alice'), (2, 'Bob'), (3, 'Catherine')]
columns = ['id', 'name']
df1 = spark.createDataFrame(data, columns)
df2 = spark.createDataFrame(data, columns)
# Cache DataFrames
df1.cache()
df2.cache()
# Add DataFrames to the list
cached_dfs.append(df1)
cached_dfs.append(df2)
# Perform some actions
df1.show()
df2.show()
# Un-persist all DataFrames
for df in cached_dfs:
df.unpersist()
Output:
+---+---------+
| id| name|
+---+---------+
| 1| Alice|
| 2| Bob|
| 3|Catherine|
+---+---------+
+---+---------+
| id| name|
+---+---------+
| 1| Alice|
| 2| Bob|
| 3|Catherine|
+---+---------+
Method 3: Using GC to Un-persist Automatically
Starting with Spark 2.3, there is a `spark.cleaner.referenceTracking.cleanCheckpoints` configuration that helps in automatic un-persisting of DataFrames when they are garbage collected. This method leverages the cleaner to automatically free up resources. However, this method is less predictable as it relies on JVM garbage collection.
spark.conf.set("spark.cleaner.referenceTracking.cleanCheckpoints", "true")
Note: Ensure that you have a solid understanding of how and when garbage collection occurs in the JVM to use this method effectively.
Conclusion
Un-persisting DataFrames is a crucial step in managing resources efficiently in Apache Spark. While un-persisting individual DataFrames is straightforward, managing multiple DataFrames can be streamlined using methods like the Spark catalog, manual tracking, or automatic garbage collection. Choose the method that best fits your application’s needs and complexity.