Why does Dataset.unpersist cascade to all dependent cached Datasets?

user_1362926 0
Viewed 578

I am using spark 2.3.2. For my use case, I'm caching first dataframe and then second dataframe.

Trying to replicate the same.

scala> val df = spark.range(1, 1000000).withColumn("rand", (rand * 100).cast("int")).cache
df: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [id: bigint, rand: int]

scala> df.count
res0: Long = 999999                                                             

scala> val aggDf = df.groupBy("rand").agg(count("id") as "count").cache
aggDf: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [rand: int, count: bigint]

scala> aggDf.count
res1: Long = 100

As use can see in below image, There are two RDD's for each dataframe.

enter image description here

Now, When I'm going to unpersist my first dataframe, spark is unpersisting both. 

df.unpersist()

Trying to understand this weird behaviour, Why spark is unpersisting both dataframe instead of first?
Am I missing something?

apache-spark-sql apache-spark
1 Answers

Quoting SPARK-21478 Unpersist a DF also unpersists related DFs:

This is by design. We do not want to use the invalid cached data.

The current cache design need to ensure the query correctness. If you want to keep the cached data, even if the data is stale. You need to materialize it by saving it as a table.

That however has been changed in 2.4.0 in SPARK-24596 Non-cascading Cache Invalidation:

When invalidating a cache, we invalid other caches dependent on this cache to ensure cached data is up to date. For example, when the underlying table has been modified or the table has been dropped itself, all caches that use this table should be invalidated or refreshed.

However, in other cases, like when user simply want to drop a cache to free up memory, we do not need to invalidate dependent caches since no underlying data has been changed. For this reason, we would like to introduce a new cache invalidation mode: the non-cascading cache invalidation.

Since you're using 2.3.2 you have to follow the recommendation to save to a table or upgrade to 2.4.0.

user_13053440
Related