JFYI, you can use pre-commit to avoid code-style errors:
https://github.com/graphframes/graphframes/blob/master/CONTRIBUTING.md#styleguides

JFYI, you can use pre-commit to avoid code-style errors: https://github.com/graphframes/graphframes/blob/master/CONTRIBUTING.md#styleguides

I didn't change a line, only removed the counts. The code-style error was introduced previously in the #459 PR.

Technically speaking, the last round of DataFrames has already been persisted before this PR, so the only difference now is that it will persist only one DataFrame (the last one). Additionally, this PR is just a small part of the changes I plan to introduce in the ConnectedComponents. I think it's fine to update the documentation, but do you mind if I update it after completing all the changes I need to make?

I think it's fine to update the documentation, but do you mind if I update it after completing all the changes I need to make?

I'm absolutely fine with it!

I totally understand your doubts, because it's not easy to explain, don't worry. I'm not 100% sure either, but this is how I think this works:

This algorithm persists intermediate results of the iteration i, executes the _calcMinNbrSum (that contains an action) to check whether the algorithm has converged or not and then, unpersists the intermediate results of the iteration i-1 (or the ones previous to the loop in the first iteration).

Once the loop has ended and the last action is performed in _calcMinNbrSum, the last resulting dataframe is obtained and persisted and then - and only then - the previous ones are unpersisted.

As you said, given that persist and unpersist are lazy operations, once you have persisted a new dataframe, you can perfectly unpersist the previous dataframes that generated it. So, I don't think an action is needed every time we persist a dataframe in order to unpersist it.

This is a super interesting topic, btw. I have in mind to write a couple of articles about how cache works in Spark.
I think I'm right, but it would be great if you can probe me wrong ... because I'd learn something new about how persist really works in Spark.

As you said, given that persist and unpersist are lazy operations, once you have persisted a new dataframe, you can perfectly unpersist the previous dataframes that generated it.

This is not what I am saying. I say "given that persist and unpersist are lazy operations, once you have persisted a new dataframe, you can" NOT "perfectly unpersist", because persisting the dataframe does not place the dataframe into the cache! When query execution occurs, the cache lookup will not return the unpersisted dataframe and the results are calculated with 0 cache hits.

it would be great if you can [prove] me wrong

I think the test that I added to my PR and the executed plans does prove you wrong, but I have gone ahead and create a minimal example showing how caching works in spark. you can add this to the ConnectedComponents test suite and play with it yourself:

  test("query planning and cache") {
    spark.conf.set("spark.sql.adaptive.enabled", "false")
    spark.catalog.clearCache()
    val df1 = spark.sql("SELECT 1 * 10 AS col1").cache()
    df1.count()

    val df2 = df1.withColumn("col2", expr("col1 * 1000"))

    df2.persist()
    df1.unpersist()
    df2.collect()
    val noCountExecutedPlan = df2.queryExecution.executedPlan

    spark.catalog.clearCache()
    val df3 = spark.sql("SELECT 1 * 10 AS col1").cache()
    df3.count()

    val df4 = df3.withColumn("col2", expr("col1 * 1000"))

    df4.persist().count()
    df3.unpersist()
    df4.collect()
    val countExecutedPlan = df4.queryExecution.executedPlan

    println("---Plan without Count---")
    println(noCountExecutedPlan)
    println("---Plan with Count---")
    println(countExecutedPlan)

  }

gives the printed output:

---Plan without Count---
InMemoryTableScan [col1#0, col2#38]
   +- InMemoryRelation [col1#0, col2#38], StorageLevel(disk, memory, deserialized, 1 replicas)
         +- *(1) Project [10 AS col1#0, 10000 AS col2#38]
            +- *(1) Scan OneRowRelation[]

---Plan with Count---
InMemoryTableScan [col1#119, col2#157]
   +- InMemoryRelation [col1#119, col2#157], StorageLevel(disk, memory, deserialized, 1 replicas)
         +- *(1) Project [col1#119, (col1#119 * 1000) AS col2#157]
            +- InMemoryTableScan [col1#119]
                  +- InMemoryRelation [col1#119], StorageLevel(disk, memory, deserialized, 1 replicas)
                        +- *(1) Project [10 AS col1#119]
                           +- *(1) Scan OneRowRelation[]

Edit: updated example to disable AQE to be easier to understand

Have you had a look at the comments in the unpersist method?

   * Mark the Dataset as non-persistent, and remove all blocks for it from memory and disk.
   * This will not un-persist any cached data that is built upon this Dataset.

So, theoretically, if we build another Dataset based on the persisted one and persist it, we could perfectly unpersist the previous one.

However ... I think you're right and I was wrong. Seems like while persist is lazy, unpersist is not.

In the following test, if persisting/unpersisting dataframes worked as I thought, only one message should be printed, instead we have two. Not only that but, as you also showed in your test, the explain plan changes after unpersisting the first dataframe. When an action is perfomed on the second dataframe before unpersisting the first one, the plan doesn't change.
THANKS FOR PROBING ME WRONG!

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.StringType

object SparkCacheExample2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .getOrCreate()

    import spark.implicits._

    // Create the first DataFrame with one row
    var firstDF = spark.range(0,1)

    firstDF = firstDF.map(x => {
      print("I'm the first DataFrame!!");
      x
    })

    // Cache and show the first DataFrame
    firstDF.cache().show()

    // Create a second DataFrame based on the first
    val secondDF = firstDF.select(firstDF("value").cast(StringType))

    // Cache the second DataFrame, unpersist the first DataFrame and show the second DataFrame
    secondDF.cache()
    secondDF.explain()
    firstDF.unpersist()
    secondDF.explain()
    secondDF.show()

    // Stop Spark session
    spark.stop()
  }
}

I think the comments of these methods are wrong or, at least, misleading.

Btw, are we sure performing a count is caching everything from the dataframes?

Before merging this PR, I'd like to ensure that we're persisting everything necessary and that Spark is not recalculating anything.

Im going to speak from experience rather than any inspection of source code in this comment.

• This will not un-persist any cached data that is built upon this Dataset.

A dataframe that has had cache or persist called is not built yet. its is only built when actions are called on it. certain actions only build part of the dataframe (show for example might not cache all of the df).

are we sure performing a count is caching everything from the dataframes?

When I have checked, count has always cached 100% of the data for me as long as the storage is available (and of course there are no executor failures.).

There's still something weird ... when I replace the map transformation with transform in my test - replacing also the "value" reference with "id"- the message gets printed only once.

Not only that, but the same behaviour is observed in the ConnectedComponents tests "single vertex" and "intermediate storage level": there are no repeated messages - per run execution - when I add the following code just after the checkpoint block inside the loop:

          System.gc() // hint Spark to clean shuffle directories
        }
        ee = ee.transform(x => {
          println(s"Iteration: $iteration")
          x
        })

        ee.persist(intermediateStorageLevel)

How can we explain that?

As I said earlier, single vertex case is not representative. I prefer the chain graph case.

I believe this is an edge case because spark is optimizing away the second child of the join because ee is an empty LocalRelation.

You ask

How can we explain that?

You are returning the same dataframe in your lambda. so the print isn't part of the query plan. pyspark docs are more helpful than scala docs at describing what the transform method does: https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.transform.html

Try this instead:

        val myUdf = udf((x: Int) => {
          println("&tk")
          x
        })
        ee = ee.withColumn(SRC, myUdf(col(SRC)))

then count the &tks that come out. It is 12 w/ count and 28 w/o when using the chain graph case.

I got mixed up and thought I was using the transform UDF (I didn't know the DataFrame had that method too). Sorry.

I've discovered how to determine whether the cache buffers of a DataFrame have been loaded or not:

spark.sharedState.cacheManager.lookupCachedData(df).get.cachedRepresentation.cacheBuilder.isCachedColumnBuffersLoaded

When no action is performed on df, this returns false. After performing some action on the DataFrame, it returns true."

We only need to execute count on the output DataFrame to achieve both goals:

• A persisted and loaded in-scope DataFrame for the caller
• A significant reduction in the number of DataFrames with redundant actions performed

Btw, what about logging a warning message to inform the user that the returned DataFrame is persisted? Something like ...

logWarn("The DataFrame returned by ConnectedComponents is persisted and loaded.")

⚠️ Please install the to ensure uploads and comments are reliably processed by Codecov.

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 89.85%. Comparing base (bc487ef) to head (788378a).
Report is 17 commits behind head on master.

❗ Your organization needs to install the Codecov GitHub app to enable full functionality.

@@            Coverage Diff             @@
##           master     #552      +/-   ##
==========================================
- Coverage   91.43%   89.85%   -1.58%     
==========================================
  Files          18       18              
  Lines         829      907      +78     
  Branches       52      118      +66     
==========================================
+ Hits          758      815      +57     
- Misses         71       92      +21     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

Im'm trying to see how we can be confident that each of these dataframes is 100% cached (rather than partially)

Edit: it does, as seen in InMemoryRelation.isCachedRDDLoaded

Well, as you mentioned, the Spark UI shows 100% is cached. Besides, because the dataframes have only a few columns ... there shouldn't be any problem.

We could replace the counts with write.format("noop").mode("overwite").save() that will assure us the 100%.

@james-willis you have permissions to approve? Can you squash and merge? Can you cut a release?

@james-willis you have permissions to approve? Can you squash and merge? Can you cut a release?

I think only you, @rjurney, have a bage "Collaborator" and the write access. All of us have only "Contributor" badges.

I am not even a contributor to this repo yet! haha

I am not even a contributor to this repo yet! haha

I liked the "yet" part 😂. I had a look at Sedona also to contribute somehow but ... maybe it's to high level for me ... 😂

It seems that @SemyonSinchenko is a Collaborator now. Perhaps we can finally get this fix merged after 6 months

@SauronShepherd Are you going to do anything else here? Or can it be merged?

I have in mind adding a few more changes, but I'll open one (or probably more) PRs. Please, go ahead with the merge. Thanks

This PR is related to #459 (maybe I shouldn't have opened a new one,). We should close both once the merge is carried out.