Does Spark eject DataFrame data from memory every time an action is executed?

chad 0
Viewed 39

I'm trying to understand how to leverage cache() to improve my performance. Since cache retains a DataFrame in memory "for reuse", it seems like i need to understand the conditions that eject the DataFrame from memory to better understand how to leverage it.

After defining transformations, I call an action, is the dataframe, after the action completes, gone from memory? This would imply that if I do execute an action on a dataframe, but I continue to do other stuff with the data, all the previous parts of the DAG, from the read to the action, will need to be re done.

Is this accurate?

pyspark apache-spark python
2 Answers
thebluephantom0

The fact is, that after an action is executed another dag is created. You can check this via SparkUI

In code you can try to identify where your dag is done and new started by looking for actions

When looking at code you can use this simple rule:

  • When function is transforming one df into another df - its not action but only lazy evaluated transformation. Even if this is join or something else which requires shuffling

  • When fuction is returning value other dan df, then you are using and action (for example count which is returning Long)

Lets take a look at this code (Its Scala but api is similar). I created this example just to show you the mechanism, this could be done better of course :)

import org.apache.spark.sql.functions.{col, lit, format_number}
import org.apache.spark.sql.DataFrame

val input = spark.read.format("csv")
                  .option("header", "true")
                  .load("dbfs:/FileStore/shared_uploads/***@gmail.com/city_temperature.csv")

val dataWithTempInCelcius: DataFrame = input
                 .withColumn("avg_temp_celcius",format_number((col("AvgTemperature").cast("integer") - lit(32)) / lit(1.8), 2).cast("Double"))

val dfAfterGrouping: DataFrame = dataWithTempInCelcius
                .filter(col("City") === "Warsaw")
                .groupBy(col("Year"), col("Month"))
                .max("avg_temp_celcius")//Not an action, we are still doing transofrmation from df to another df

val maxTemp: Row = dfAfterGrouping
                 .orderBy(col("Year"))
                .first() // <--- first is our first action, output from first is a Row and not df

dataWithTempInCelcius
                 .filter(col("City") === "Warsaw")
                 .count() // <--- count is our second action

Here you may see what is the problem. It looks like between first and second action i am doing transformation which was already done in first dag. This intermediate results of calculation was not cached, so in second dag Spark is unable to get the dataframe after filter from the memory which leads us to recomputation. Spark is going to fetch the data again, apply our filter and then calculate the count.

In SparkUI u will find two separate dags and both of them are going to read the source csv

First action on left, second on right

If you cache intermediate results after first .filter(col("City") === "Warsaw") and then use this cached DF to do grouping and count you will still find two separate dags (number of action has not changed) but this time in the plan for second dag you will find "In memory table scan" instead of read of a csv file - that means that Spark is reading data from cache

First action on left, second on right

Now you can see in memory relation in plan. There is still read csv node in the dag but as you can see, for second action its skipped (0 bytes read)

** I am using Databrics cluster with Spark 3.2, SparkUI may look different on your env

M_S0
Related