Is it efficient to cache a dataframe for a single Action Spark application in which that dataframe is referenced more than once?

Question

I am little confused with the caching mechanism of Spark.

Let's say I have a Spark application with only one action at the end of multiple transformations. In which suppose I have a dataframe A and I applied 2-3 transformation on it, creating multiple dataframes which eventually helps creating a last dataframe which is going to be saved to disk.

example :

val A=spark.read() // large size
val B=A.map()
val C=A.map()
.
.
.
val D=B.join(C)
D.save()

So do I need to cache dataframe A for performance enhancement?

Thanks in advance.

Answer 1

I think thebluephantom's answer is right.

I had faced same situation with you until today, and i also found answers only saying Spark cache() does not work on single query. And also my spark job executing single query seems not caching.

Because of them, i was also doubted for he's answer.
But he showed evidences for cache is working with a green box even he execute single query.

So, i tested 3 cases with dataframe(not RDD) like below and the results seems he is right.
And execution plan is also changed (more simple and use InMemoryRelation, please see the below).

1. without cache
2. using cache
3. using cache with calling unpersist before action

without cache

example

val A = spark.read.format().load()

val B = A.where(cond1).select(columns1)
val C = A.where(cond2).select(columns2)

val D = B.join(C)

D.save()

DAG for my case

This is a bit more complicated than example.

This DAG is messy even though there is no complicated execution. And you can see the scan is occured 4 times.

with cache

example

val A = spark.read.format().load().cache()  // cache will be working

val B = A.where(cond1).select(columns1)
val C = A.where(cond2).select(columns2)

val D = B.join(C)

D.save()

This will cache A, even single query.
You can see DAG that read InMemoryTableScan twice.

DAG for my case

with cache and unpersist before action

val A = spark.read.format().load().cache()

val B = A.where(cond1).select(columns1)
val C = A.where(cond2).select(columns2)

/* I thought A will not be needed anymore */
A.unpersist()

val D = B.join(C)

D.save()

This code will not cache A dataframe, because it was unset cache flag before starting action. (D.save()) So, this will result in exactly same with first case (without cache).

Important thing is unpersist() must be written after action(after D.save()). But when i ask some people in my company, many of them used like case 3 and didn't know about this.

I think that's why many people misunderstand cache is not working on single query.

cache and unpersist should be like below

val A = spark.read.format().load().cache()

val B = A.where(cond1).select(columns1)
val C = A.where(cond2).select(columns2)

val D = B.join(C)

D.save()

/* unpersist must be after action */
A.unpersist()

This result exactly same with case 2 (with cache, but unpersist after D.save()).

So. I suggest try cache like thebluephantom's answer.
If i present any incorrection. please note that.

Thanks to thebluephantom's for solving my problem.

Answer 2

Yes, you are correct.

You should cache A as it used for B & C as input. The DAG visualization would show the extent of reuse or going back to source (in this case). If you have a noisy cluster, some spilling to disk could occur.

See also top answer here (Why) do we need to call cache or persist on a RDD

However, I was looking for skipped stages, silly me. But something else shows as per below.

The following code akin to your own code:

val aa = spark.sparkContext.textFile("/FileStore/tables/filter_words.txt")//.cache
val a = aa.flatMap(x => x.split(" ")).map(_.trim) 
val b=a.map(x => (x,1)) 
val c=a.map(x => (x,2)) 
val d=b.join(c)
d.count

Looking at UI with .cache

and without .cache

QED: SO, .cache has benefit. It would not make sense otherwise. Also, 2 reads could lead to different results in some cases.