getPersistentRDDs returns Map of cached RDDs and DataFrames in Spark 2.2.0, but in Spark 2.4.7 - it returns Map of cached RDDs only

user_7618482 0
Viewed 189

If cache RDD and DataFrame in Spark version 2.2.0 getPersistentRDDs returns Map size 2:

scala> val rdd = sc.parallelize(Seq(1))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> val df = sc.parallelize(Seq(2)).toDF
df: org.apache.spark.sql.DataFrame = [value: int]

scala> spark.sparkContext.getPersistentRDDs
res0: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map()

scala> df.cache
res1: df.type = [value: int]

scala> spark.sparkContext.getPersistentRDDs
res2: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] =
Map(4 -> *SerializeFromObject [input[0, int, false] AS value#2]
+- Scan ExternalRDDScan[obj#1]
 MapPartitionsRDD[4] at cache at <console>:27)

scala> rdd.cache
res3: rdd.type = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> spark.sparkContext.getPersistentRDDs
res4: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] =
Map(0 -> ParallelCollectionRDD[0] at parallelize at <console>:24, 4 -> *SerializeFromObject [input[0, int, false] AS value#2]
+- Scan ExternalRDDScan[obj#1]
 MapPartitionsRDD[4] at cache at <console>:27)

but in Spark version 2.4.7 getPersistentRDDs returns Map size 1

...
scala> spark.sparkContext.getPersistentRDDs
res4: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map(0 -> ParallelCollectionRDD[0] at parallelize at <console>:24)

And the question how to get all cached objects not just RDDs and what suddenly happened to the method that began to behave differently?

rdd apache-spark scala
1 Answers

The dataframe is not actually cached in memory because no action has been performed on the dataframe yet, so it's actually fair to exclude it from the result of getPersistentRDDs. I think the behaviour in the later version is actually desirable. But once you have done something on the dataframe, it will be cached, and it will appear in the result of getPersistentRDDs, as shown below:

scala> val df = sc.parallelize(Seq(2)).toDF
df: org.apache.spark.sql.DataFrame = [value: int]

scala> sc.getPersistentRDDs
res0: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map()

scala> df.cache
res1: df.type = [value: int]

scala> sc.getPersistentRDDs
res2: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map()

scala> df.count()
res3: Long = 1

scala> sc.getPersistentRDDs
res4: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] =
Map(3 -> *(1) SerializeFromObject [input[0, int, false] AS value#2]
+- Scan[obj#1]
 MapPartitionsRDD[3] at count at <console>:26)
user_141657300
Related