Spark list all cached RDD names and unpersist
PySparkers: getPersistentRDDs isn't yet implemented in Python, so unpersist your RDDs by dipping into Java:
for (id, rdd) in spark.sparkContext._jsc.getPersistentRDDs().items():
rdd.unpersist()
Scala generic way of doing this ... loop through spark context get all persistent RDDs and unpersist
.
I will use this at the end of a driver.
for ( (id,rdd) <- sparkSession.sparkContext.getPersistentRDDs ) {
log.info("Unexpected cached RDD " + id)
rdd.unpersist()
}
Java Generic way of doing this ... where jsc is JavaSparkContext
if (jsc != null) {
Map<Integer, JavaRDD<?>> persistentRDDS = jsc.getPersistentRDDs();
// using for-each loop for iteration over Map.entrySet()
for (Map.Entry<Integer, JavaRDD<?>> entry : persistentRDDS.entrySet()) {
LOG.info("Key = " + entry.getKey() +
", un persisting cached RDD = " + entry.getValue().unpersist());
}
}
Another short form of unpersist
in java with out knowing rdd names are :
Map<Integer, JavaRDD<?>> persistentRDDS = jsc.getPersistentRDDs();
persistentRDDS.values().forEach(JavaRDD::unpersist);
@Dikei's answer is actually correct but I believe what you are looking for is sc.getPersistentRDDs
:
scala> val rdd1 = sc.makeRDD(1 to 100)
# rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at makeRDD at <console>:27
scala> val rdd2 = sc.makeRDD(10 to 1000)
# rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at makeRDD at <console>:27
scala> rdd2.cache.setName("rdd_2")
# res0: rdd2.type = rdd_2 ParallelCollectionRDD[1] at makeRDD at <console>:27
scala> sc.getPersistentRDDs
# res1: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map(1 -> rdd_2 ParallelCollectionRDD[1] at makeRDD at <console>:27)
scala> rdd1.cache.setName("foo")
# res2: rdd1.type = foo ParallelCollectionRDD[0] at makeRDD at <console>:27
scala> sc.getPersistentRDDs
# res3: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map(1 -> rdd_2 ParallelCollectionRDD[1] at makeRDD at <console>:27, 0 -> foo ParallelCollectionRDD[0] at makeRDD at <console>:27)
Now let's add another RDD
and name it as well :
scala> rdd3.setName("bar")
# res4: rdd3.type = bar ParallelCollectionRDD[2] at makeRDD at <console>:27
scala> sc.getPersistentRDDs
# res5: scala.collection.Map[Int,org.apache.spark.rdd.RDD[_]] = Map(1 -> rdd_2 ParallelCollectionRDD[1] at makeRDD at <console>:27, 0 -> foo ParallelCollectionRDD[0] at makeRDD at <console>:27)
We noticed that actually it isn't persisted.