You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sr...@apache.org on 2019/02/02 00:30:18 UTC
[spark] branch master updated: [SPARK-26771][CORE][GRAPHX] Make
.unpersist(), .destroy() consistently non-blocking by default
This is an automated email from the ASF dual-hosted git repository.
srowen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 8171b156 [SPARK-26771][CORE][GRAPHX] Make .unpersist(), .destroy() consistently non-blocking by default
8171b156 is described below
commit 8171b156ebf25a582318c71222716b5c52c8bbb3
Author: Sean Owen <se...@databricks.com>
AuthorDate: Fri Feb 1 18:29:55 2019 -0600
[SPARK-26771][CORE][GRAPHX] Make .unpersist(), .destroy() consistently non-blocking by default
## What changes were proposed in this pull request?
Make .unpersist(), .destroy() non-blocking by default and adjust callers to request blocking only where important.
This also adds an optional blocking argument to Pyspark's RDD.unpersist(), which never had one.
## How was this patch tested?
Existing tests.
Closes #23685 from srowen/SPARK-26771.
Authored-by: Sean Owen <se...@databricks.com>
Signed-off-by: Sean Owen <se...@databricks.com>
---
.../main/scala/org/apache/spark/MapOutputTracker.scala | 2 +-
.../src/main/scala/org/apache/spark/SparkContext.scala | 2 +-
.../scala/org/apache/spark/broadcast/Broadcast.scala | 3 +--
core/src/main/scala/org/apache/spark/rdd/RDD.scala | 6 +++---
.../spark/rdd/util/PeriodicRDDCheckpointer.scala | 2 +-
.../test/scala/org/apache/spark/DistributedSuite.scala | 2 +-
.../test/scala/org/apache/spark/UnpersistSuite.scala | 2 +-
.../org/apache/spark/broadcast/BroadcastSuite.scala | 4 ++--
.../scala/org/apache/spark/ui/UISeleniumSuite.scala | 4 ++--
docs/graphx-programming-guide.md | 2 +-
docs/rdd-programming-guide.md | 9 ++++++++-
.../spark/examples/mllib/BinaryClassification.scala | 2 +-
.../spark/examples/mllib/DecisionTreeRunner.scala | 2 +-
.../apache/spark/examples/mllib/LinearRegression.scala | 2 +-
.../org/apache/spark/examples/mllib/MovieLensALS.scala | 2 +-
.../src/main/scala/org/apache/spark/graphx/Graph.scala | 8 ++++++--
.../main/scala/org/apache/spark/graphx/Pregel.scala | 6 +++---
.../org/apache/spark/graphx/impl/EdgeRDDImpl.scala | 2 +-
.../scala/org/apache/spark/graphx/impl/GraphImpl.scala | 4 ++--
.../org/apache/spark/graphx/impl/VertexRDDImpl.scala | 2 +-
.../scala/org/apache/spark/graphx/lib/PageRank.scala | 10 +++++-----
.../spark/graphx/lib/StronglyConnectedComponents.scala | 2 +-
.../spark/graphx/util/PeriodicGraphCheckpointer.scala | 2 +-
.../scala/org/apache/spark/graphx/GraphSuite.scala | 10 +++++-----
.../org/apache/spark/ml/classification/LinearSVC.scala | 2 +-
.../spark/ml/classification/LogisticRegression.scala | 2 +-
.../apache/spark/ml/clustering/GaussianMixture.scala | 6 +++---
.../apache/spark/ml/optim/loss/RDDLossFunction.scala | 2 +-
.../spark/ml/regression/AFTSurvivalRegression.scala | 4 ++--
.../apache/spark/ml/regression/LinearRegression.scala | 4 ++--
.../spark/ml/tree/impl/GradientBoostedTrees.scala | 2 +-
.../org/apache/spark/ml/tree/impl/NodeIdCache.scala | 6 +++---
.../apache/spark/mllib/api/python/PythonMLLibAPI.scala | 16 ++++++++--------
.../spark/mllib/clustering/BisectingKMeans.scala | 8 ++++----
.../spark/mllib/clustering/GaussianMixture.scala | 2 +-
.../org/apache/spark/mllib/clustering/KMeans.scala | 12 ++++++------
.../apache/spark/mllib/clustering/KMeansModel.scala | 2 +-
.../org/apache/spark/mllib/clustering/LDAModel.scala | 2 +-
.../apache/spark/mllib/clustering/LDAOptimizer.scala | 2 +-
.../org/apache/spark/mllib/feature/Word2Vec.scala | 10 +++++-----
.../scala/org/apache/spark/mllib/fpm/PrefixSpan.scala | 2 +-
.../spark/mllib/optimization/GradientDescent.scala | 2 +-
.../org/apache/spark/mllib/optimization/LBFGS.scala | 2 +-
.../mllib/regression/GeneralizedLinearAlgorithm.scala | 2 +-
.../spark/mllib/tree/model/treeEnsembleModels.scala | 2 +-
.../spark/ml/classification/LinearSVCSuite.scala | 4 ++--
.../ml/classification/LogisticRegressionSuite.scala | 6 +++---
python/pyspark/broadcast.py | 11 +++++++----
python/pyspark/rdd.py | 8 ++++++--
python/pyspark/tests/test_rdd.py | 2 +-
.../org/apache/spark/sql/DataFrameFunctionsSuite.scala | 2 +-
.../scala/org/apache/spark/sql/DatasetCacheSuite.scala | 18 +++++++++---------
.../columnar/InMemoryColumnarQuerySuite.scala | 6 +++---
.../org/apache/spark/sql/hive/CachedTableSuite.scala | 2 +-
.../org/apache/spark/streaming/dstream/DStream.scala | 2 +-
55 files changed, 131 insertions(+), 114 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
index a8b8e96..803703f 100644
--- a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
+++ b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
@@ -201,7 +201,7 @@ private class ShuffleStatus(numPartitions: Int) {
Utils.tryLogNonFatalError {
// Use `blocking = false` so that this operation doesn't hang while trying to send cleanup
// RPCs to dead executors.
- cachedSerializedBroadcast.destroy(blocking = false)
+ cachedSerializedBroadcast.destroy()
}
cachedSerializedBroadcast = null
}
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index 6273601..6042420 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -1753,7 +1753,7 @@ class SparkContext(config: SparkConf) extends Logging {
/**
* Unpersist an RDD from memory and/or disk storage
*/
- private[spark] def unpersistRDD(rddId: Int, blocking: Boolean = true) {
+ private[spark] def unpersistRDD(rddId: Int, blocking: Boolean) {
env.blockManager.master.removeRdd(rddId, blocking)
persistentRdds.remove(rddId)
listenerBus.post(SparkListenerUnpersistRDD(rddId))
diff --git a/core/src/main/scala/org/apache/spark/broadcast/Broadcast.scala b/core/src/main/scala/org/apache/spark/broadcast/Broadcast.scala
index 24d953e..0e81ad1 100644
--- a/core/src/main/scala/org/apache/spark/broadcast/Broadcast.scala
+++ b/core/src/main/scala/org/apache/spark/broadcast/Broadcast.scala
@@ -92,10 +92,9 @@ abstract class Broadcast[T: ClassTag](val id: Long) extends Serializable with Lo
/**
* Destroy all data and metadata related to this broadcast variable. Use this with caution;
* once a broadcast variable has been destroyed, it cannot be used again.
- * This method blocks until destroy has completed
*/
def destroy() {
- destroy(blocking = true)
+ destroy(blocking = false)
}
/**
diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
index 954582f..d39f418 100644
--- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
@@ -212,11 +212,11 @@ abstract class RDD[T: ClassTag](
/**
* Mark the RDD as non-persistent, and remove all blocks for it from memory and disk.
*
- * @param blocking Whether to block until all blocks are deleted.
+ * @param blocking Whether to block until all blocks are deleted (default: false)
* @return This RDD.
*/
- def unpersist(blocking: Boolean = true): this.type = {
- logInfo("Removing RDD " + id + " from persistence list")
+ def unpersist(blocking: Boolean = false): this.type = {
+ logInfo(s"Removing RDD $id from persistence list")
sc.unpersistRDD(id, blocking)
storageLevel = StorageLevel.NONE
this
diff --git a/core/src/main/scala/org/apache/spark/rdd/util/PeriodicRDDCheckpointer.scala b/core/src/main/scala/org/apache/spark/rdd/util/PeriodicRDDCheckpointer.scala
index 5e181a9..4a61069 100644
--- a/core/src/main/scala/org/apache/spark/rdd/util/PeriodicRDDCheckpointer.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/util/PeriodicRDDCheckpointer.scala
@@ -89,7 +89,7 @@ private[spark] class PeriodicRDDCheckpointer[T](
}
}
- override protected def unpersist(data: RDD[T]): Unit = data.unpersist(blocking = false)
+ override protected def unpersist(data: RDD[T]): Unit = data.unpersist()
override protected def getCheckpointFiles(data: RDD[T]): Iterable[String] = {
data.getCheckpointFile.map(x => x)
diff --git a/core/src/test/scala/org/apache/spark/DistributedSuite.scala b/core/src/test/scala/org/apache/spark/DistributedSuite.scala
index 17ffc7b..c0c8899 100644
--- a/core/src/test/scala/org/apache/spark/DistributedSuite.scala
+++ b/core/src/test/scala/org/apache/spark/DistributedSuite.scala
@@ -323,7 +323,7 @@ class DistributedSuite extends SparkFunSuite with Matchers with LocalSparkContex
data.persist(StorageLevel.MEMORY_ONLY_2)
data.count
assert(sc.persistentRdds.isEmpty === false)
- data.unpersist()
+ data.unpersist(blocking = true)
assert(sc.persistentRdds.isEmpty === true)
failAfter(Span(3000, Millis)) {
diff --git a/core/src/test/scala/org/apache/spark/UnpersistSuite.scala b/core/src/test/scala/org/apache/spark/UnpersistSuite.scala
index b58a3eb..2432d6d 100644
--- a/core/src/test/scala/org/apache/spark/UnpersistSuite.scala
+++ b/core/src/test/scala/org/apache/spark/UnpersistSuite.scala
@@ -30,7 +30,7 @@ class UnpersistSuite extends SparkFunSuite with LocalSparkContext with TimeLimit
val rdd = sc.makeRDD(Array(1, 2, 3, 4), 2).cache()
rdd.count
assert(sc.persistentRdds.isEmpty === false)
- rdd.unpersist()
+ rdd.unpersist(blocking = true)
assert(sc.persistentRdds.isEmpty === true)
failAfter(Span(3000, Millis)) {
diff --git a/core/src/test/scala/org/apache/spark/broadcast/BroadcastSuite.scala b/core/src/test/scala/org/apache/spark/broadcast/BroadcastSuite.scala
index 18ec60d..66b2f48 100644
--- a/core/src/test/scala/org/apache/spark/broadcast/BroadcastSuite.scala
+++ b/core/src/test/scala/org/apache/spark/broadcast/BroadcastSuite.scala
@@ -296,7 +296,7 @@ class BroadcastSuite extends SparkFunSuite with LocalSparkContext with Encryptio
// Using this variable on the executors crashes them, which hangs the test.
// Instead, crash the driver by directly accessing the broadcast value.
intercept[SparkException] { broadcast.value }
- intercept[SparkException] { broadcast.unpersist() }
+ intercept[SparkException] { broadcast.unpersist(blocking = true) }
intercept[SparkException] { broadcast.destroy(blocking = true) }
} else {
val results = sc.parallelize(1 to partitions, partitions).map(x => (x, broadcast.value.sum))
@@ -309,7 +309,7 @@ package object testPackage extends Assertions {
def runCallSiteTest(sc: SparkContext) {
val broadcast = sc.broadcast(Array(1, 2, 3, 4))
- broadcast.destroy()
+ broadcast.destroy(blocking = true)
val thrown = intercept[SparkException] { broadcast.value }
assert(thrown.getMessage.contains("BroadcastSuite.scala"))
}
diff --git a/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala b/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
index b184b74..cdc7185 100644
--- a/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/UISeleniumSuite.scala
@@ -137,7 +137,7 @@ class UISeleniumSuite extends SparkFunSuite with WebBrowser with Matchers with B
val rddJson = getJson(ui, "storage/rdd/0")
(rddJson \ "storageLevel").extract[String] should be (StorageLevels.DISK_ONLY.description)
- rdd.unpersist()
+ rdd.unpersist(blocking = true)
rdd.persist(StorageLevels.MEMORY_ONLY).count()
eventually(timeout(5 seconds), interval(50 milliseconds)) {
goToUi(ui, "/storage")
@@ -172,7 +172,7 @@ class UISeleniumSuite extends SparkFunSuite with WebBrowser with Matchers with B
dist0.onHeapMemoryUsed should not be (Some(0L))
dist0.offHeapMemoryUsed should be (Some(0L))
- rdd.unpersist()
+ rdd.unpersist(blocking = true)
rdd.persist(StorageLevels.OFF_HEAP).count()
val updatedStorageJson1 = getJson(ui, "storage/rdd")
updatedStorageJson1.children.length should be (1)
diff --git a/docs/graphx-programming-guide.md b/docs/graphx-programming-guide.md
index ecedeaf..6a0be7b 100644
--- a/docs/graphx-programming-guide.md
+++ b/docs/graphx-programming-guide.md
@@ -283,7 +283,7 @@ class Graph[VD, ED] {
// Functions for caching graphs ==================================================================
def persist(newLevel: StorageLevel = StorageLevel.MEMORY_ONLY): Graph[VD, ED]
def cache(): Graph[VD, ED]
- def unpersistVertices(blocking: Boolean = true): Graph[VD, ED]
+ def unpersistVertices(blocking: Boolean = false): Graph[VD, ED]
// Change the partitioning heuristic ============================================================
def partitionBy(partitionStrategy: PartitionStrategy): Graph[VD, ED]
// Transform vertex and edge attributes ==========================================================
diff --git a/docs/rdd-programming-guide.md b/docs/rdd-programming-guide.md
index 308a8ea..7f42765 100644
--- a/docs/rdd-programming-guide.md
+++ b/docs/rdd-programming-guide.md
@@ -1270,7 +1270,8 @@ waiting to recompute a lost partition.
Spark automatically monitors cache usage on each node and drops out old data partitions in a
least-recently-used (LRU) fashion. If you would like to manually remove an RDD instead of waiting for
-it to fall out of the cache, use the `RDD.unpersist()` method.
+it to fall out of the cache, use the `RDD.unpersist()` method. Note that this method does not
+block by default. To block until resources are freed, specify `blocking=true` when calling this method.
# Shared Variables
@@ -1342,6 +1343,12 @@ run on the cluster so that `v` is not shipped to the nodes more than once. In ad
`v` should not be modified after it is broadcast in order to ensure that all nodes get the same
value of the broadcast variable (e.g. if the variable is shipped to a new node later).
+To release the resources that the broadcast variable copied onto executors, call `.unpersist()`.
+If the broadcast is used again afterwards, it will be re-broadcast. To permanently release all
+resources used by the broadcast variable, call `.destroy()`. The broadcast variable can't be used
+after that. Note that these methods do not block by default. To block until resources are freed,
+specify `blocking=true` when calling them.
+
## Accumulators
Accumulators are variables that are only "added" to through an associative and commutative operation and can
diff --git a/examples/src/main/scala/org/apache/spark/examples/mllib/BinaryClassification.scala b/examples/src/main/scala/org/apache/spark/examples/mllib/BinaryClassification.scala
index a1a5b59..e3cc1d9 100644
--- a/examples/src/main/scala/org/apache/spark/examples/mllib/BinaryClassification.scala
+++ b/examples/src/main/scala/org/apache/spark/examples/mllib/BinaryClassification.scala
@@ -117,7 +117,7 @@ object BinaryClassification {
val numTest = test.count()
println(s"Training: $numTraining, test: $numTest.")
- examples.unpersist(blocking = false)
+ examples.unpersist()
val updater = params.regType match {
case L1 => new L1Updater()
diff --git a/examples/src/main/scala/org/apache/spark/examples/mllib/DecisionTreeRunner.scala b/examples/src/main/scala/org/apache/spark/examples/mllib/DecisionTreeRunner.scala
index fa47e12..dd363f5 100644
--- a/examples/src/main/scala/org/apache/spark/examples/mllib/DecisionTreeRunner.scala
+++ b/examples/src/main/scala/org/apache/spark/examples/mllib/DecisionTreeRunner.scala
@@ -247,7 +247,7 @@ object DecisionTreeRunner {
val numTest = test.count()
println(s"numTraining = $numTraining, numTest = $numTest.")
- examples.unpersist(blocking = false)
+ examples.unpersist()
(training, test, numClasses)
}
diff --git a/examples/src/main/scala/org/apache/spark/examples/mllib/LinearRegression.scala b/examples/src/main/scala/org/apache/spark/examples/mllib/LinearRegression.scala
index 86aec36..03222b1 100644
--- a/examples/src/main/scala/org/apache/spark/examples/mllib/LinearRegression.scala
+++ b/examples/src/main/scala/org/apache/spark/examples/mllib/LinearRegression.scala
@@ -104,7 +104,7 @@ object LinearRegression {
val numTest = test.count()
println(s"Training: $numTraining, test: $numTest.")
- examples.unpersist(blocking = false)
+ examples.unpersist()
val updater = params.regType match {
case NONE => new SimpleUpdater()
diff --git a/examples/src/main/scala/org/apache/spark/examples/mllib/MovieLensALS.scala b/examples/src/main/scala/org/apache/spark/examples/mllib/MovieLensALS.scala
index 9bd6927..fd81015 100644
--- a/examples/src/main/scala/org/apache/spark/examples/mllib/MovieLensALS.scala
+++ b/examples/src/main/scala/org/apache/spark/examples/mllib/MovieLensALS.scala
@@ -155,7 +155,7 @@ object MovieLensALS {
val numTest = test.count()
println(s"Training: $numTraining, test: $numTest.")
- ratings.unpersist(blocking = false)
+ ratings.unpersist()
val model = new ALS()
.setRank(params.rank)
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/Graph.scala b/graphx/src/main/scala/org/apache/spark/graphx/Graph.scala
index b3a3420..de94f5b 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/Graph.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/Graph.scala
@@ -119,16 +119,20 @@ abstract class Graph[VD: ClassTag, ED: ClassTag] protected () extends Serializab
/**
* Uncaches both vertices and edges of this graph. This is useful in iterative algorithms that
* build a new graph in each iteration.
+ *
+ * @param blocking Whether to block until all data is unpersisted (default: false)
*/
- def unpersist(blocking: Boolean = true): Graph[VD, ED]
+ def unpersist(blocking: Boolean = false): Graph[VD, ED]
/**
* Uncaches only the vertices of this graph, leaving the edges alone. This is useful in iterative
* algorithms that modify the vertex attributes but reuse the edges. This method can be used to
* uncache the vertex attributes of previous iterations once they are no longer needed, improving
* GC performance.
+ *
+ * @param blocking Whether to block until all data is unpersisted (default: false)
*/
- def unpersistVertices(blocking: Boolean = true): Graph[VD, ED]
+ def unpersistVertices(blocking: Boolean = false): Graph[VD, ED]
/**
* Repartitions the edges in the graph according to `partitionStrategy`.
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala b/graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala
index 755c6fe..75248d5 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/Pregel.scala
@@ -163,9 +163,9 @@ object Pregel extends Logging {
logInfo("Pregel finished iteration " + i)
// Unpersist the RDDs hidden by newly-materialized RDDs
- oldMessages.unpersist(blocking = false)
- prevG.unpersistVertices(blocking = false)
- prevG.edges.unpersist(blocking = false)
+ oldMessages.unpersist()
+ prevG.unpersistVertices()
+ prevG.edges.unpersist()
// count the iteration
i += 1
}
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/impl/EdgeRDDImpl.scala b/graphx/src/main/scala/org/apache/spark/graphx/impl/EdgeRDDImpl.scala
index eb8abd1..2b741c3 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/impl/EdgeRDDImpl.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/impl/EdgeRDDImpl.scala
@@ -58,7 +58,7 @@ class EdgeRDDImpl[ED: ClassTag, VD: ClassTag] private[graphx] (
this
}
- override def unpersist(blocking: Boolean = true): this.type = {
+ override def unpersist(blocking: Boolean = false): this.type = {
partitionsRDD.unpersist(blocking)
this
}
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala b/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala
index 34e1253..0a97ab4 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/impl/GraphImpl.scala
@@ -78,13 +78,13 @@ class GraphImpl[VD: ClassTag, ED: ClassTag] protected (
}
}
- override def unpersist(blocking: Boolean = true): Graph[VD, ED] = {
+ override def unpersist(blocking: Boolean = false): Graph[VD, ED] = {
unpersistVertices(blocking)
replicatedVertexView.edges.unpersist(blocking)
this
}
- override def unpersistVertices(blocking: Boolean = true): Graph[VD, ED] = {
+ override def unpersistVertices(blocking: Boolean = false): Graph[VD, ED] = {
vertices.unpersist(blocking)
// TODO: unpersist the replicated vertices in `replicatedVertexView` but leave the edges alone
this
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/impl/VertexRDDImpl.scala b/graphx/src/main/scala/org/apache/spark/graphx/impl/VertexRDDImpl.scala
index 2da9762..915cf8b 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/impl/VertexRDDImpl.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/impl/VertexRDDImpl.scala
@@ -58,7 +58,7 @@ class VertexRDDImpl[VD] private[graphx] (
this
}
- override def unpersist(blocking: Boolean = true): this.type = {
+ override def unpersist(blocking: Boolean = false): this.type = {
partitionsRDD.unpersist(blocking)
this
}
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/lib/PageRank.scala b/graphx/src/main/scala/org/apache/spark/graphx/lib/PageRank.scala
index 1305c05..105ad57 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/lib/PageRank.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/lib/PageRank.scala
@@ -156,8 +156,8 @@ object PageRank extends Logging {
rankGraph.edges.foreachPartition(x => {}) // also materializes rankGraph.vertices
logInfo(s"PageRank finished iteration $iteration.")
- prevRankGraph.vertices.unpersist(false)
- prevRankGraph.edges.unpersist(false)
+ prevRankGraph.vertices.unpersist()
+ prevRankGraph.edges.unpersist()
iteration += 1
}
@@ -234,9 +234,9 @@ object PageRank extends Logging {
popActivations +:+ resetActivations
}.cache()
- rankGraph.edges.foreachPartition(x => {}) // also materializes rankGraph.vertices
- prevRankGraph.vertices.unpersist(false)
- prevRankGraph.edges.unpersist(false)
+ rankGraph.edges.foreachPartition(_ => {}) // also materializes rankGraph.vertices
+ prevRankGraph.vertices.unpersist()
+ prevRankGraph.edges.unpersist()
logInfo(s"Parallel Personalized PageRank finished iteration $i.")
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/lib/StronglyConnectedComponents.scala b/graphx/src/main/scala/org/apache/spark/graphx/lib/StronglyConnectedComponents.scala
index e4f80ffc..3393ea0 100755
--- a/graphx/src/main/scala/org/apache/spark/graphx/lib/StronglyConnectedComponents.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/lib/StronglyConnectedComponents.scala
@@ -72,7 +72,7 @@ object StronglyConnectedComponents {
sccGraph.vertices.count()
sccGraph.edges.count()
// sccGraph materialized so, unpersist can be done on previous
- prevSccGraph.unpersist(blocking = false)
+ prevSccGraph.unpersist()
prevSccGraph = sccGraph
// only keep vertices that are not final
diff --git a/graphx/src/main/scala/org/apache/spark/graphx/util/PeriodicGraphCheckpointer.scala b/graphx/src/main/scala/org/apache/spark/graphx/util/PeriodicGraphCheckpointer.scala
index 539b66f..3fbeeb4 100644
--- a/graphx/src/main/scala/org/apache/spark/graphx/util/PeriodicGraphCheckpointer.scala
+++ b/graphx/src/main/scala/org/apache/spark/graphx/util/PeriodicGraphCheckpointer.scala
@@ -98,7 +98,7 @@ private[spark] class PeriodicGraphCheckpointer[VD, ED](
}
}
- override protected def unpersist(data: Graph[VD, ED]): Unit = data.unpersist(blocking = false)
+ override protected def unpersist(data: Graph[VD, ED]): Unit = data.unpersist()
override protected def getCheckpointFiles(data: Graph[VD, ED]): Iterable[String] = {
data.getCheckpointFiles
diff --git a/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala b/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
index 88b59a3..dede3b5 100644
--- a/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
+++ b/graphx/src/test/scala/org/apache/spark/graphx/GraphSuite.scala
@@ -396,11 +396,11 @@ class GraphSuite extends SparkFunSuite with LocalSparkContext {
val g = g0.partitionBy(PartitionStrategy.EdgePartition2D, 2)
val cc = g.connectedComponents()
assert(sc.getPersistentRDDs.nonEmpty)
- cc.unpersist()
- g.unpersist()
- g0.unpersist()
- vert.unpersist()
- edges.unpersist()
+ cc.unpersist(blocking = true)
+ g.unpersist(blocking = true)
+ g0.unpersist(blocking = true)
+ vert.unpersist(blocking = true)
+ edges.unpersist(blocking = true)
assert(sc.getPersistentRDDs.isEmpty)
}
}
diff --git a/mllib/src/main/scala/org/apache/spark/ml/classification/LinearSVC.scala b/mllib/src/main/scala/org/apache/spark/ml/classification/LinearSVC.scala
index ff801ab..f017605 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/classification/LinearSVC.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/classification/LinearSVC.scala
@@ -248,7 +248,7 @@ class LinearSVC @Since("2.2.0") (
scaledObjectiveHistory += state.adjustedValue
}
- bcFeaturesStd.destroy(blocking = false)
+ bcFeaturesStd.destroy()
if (state == null) {
val msg = s"${optimizer.getClass.getName} failed."
instr.logError(msg)
diff --git a/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala b/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
index f2a5c11..c0f2366 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/classification/LogisticRegression.scala
@@ -815,7 +815,7 @@ class LogisticRegression @Since("1.2.0") (
state = states.next()
arrayBuilder += state.adjustedValue
}
- bcFeaturesStd.destroy(blocking = false)
+ bcFeaturesStd.destroy()
if (state == null) {
val msg = s"${optimizer.getClass.getName} failed."
diff --git a/mllib/src/main/scala/org/apache/spark/ml/clustering/GaussianMixture.scala b/mllib/src/main/scala/org/apache/spark/ml/clustering/GaussianMixture.scala
index bb10b32..c27ba55 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/clustering/GaussianMixture.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/clustering/GaussianMixture.scala
@@ -369,8 +369,8 @@ class GaussianMixture @Since("2.0.0") (
case (aggregator1, aggregator2) => aggregator1.merge(aggregator2)
})
- bcWeights.destroy(blocking = false)
- bcGaussians.destroy(blocking = false)
+ bcWeights.destroy()
+ bcGaussians.destroy()
if (iter == 0) {
val numSamples = sums.count
@@ -409,7 +409,7 @@ class GaussianMixture @Since("2.0.0") (
iter += 1
}
- instances.unpersist(false)
+ instances.unpersist()
val gaussianDists = gaussians.map { case (mean, covVec) =>
val cov = GaussianMixture.unpackUpperTriangularMatrix(numFeatures, covVec.values)
new MultivariateGaussian(mean, cov)
diff --git a/mllib/src/main/scala/org/apache/spark/ml/optim/loss/RDDLossFunction.scala b/mllib/src/main/scala/org/apache/spark/ml/optim/loss/RDDLossFunction.scala
index 387f7c5b..84977a6 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/optim/loss/RDDLossFunction.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/optim/loss/RDDLossFunction.scala
@@ -65,7 +65,7 @@ private[ml] class RDDLossFunction[
BLAS.axpy(1.0, regGradient, gradient)
regLoss
}.getOrElse(0.0)
- bcCoefficients.destroy(blocking = false)
+ bcCoefficients.destroy()
(newAgg.loss + regLoss, gradient.asBreeze.toDenseVector)
}
}
diff --git a/mllib/src/main/scala/org/apache/spark/ml/regression/AFTSurvivalRegression.scala b/mllib/src/main/scala/org/apache/spark/ml/regression/AFTSurvivalRegression.scala
index 8d6e366..067dfa4 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/regression/AFTSurvivalRegression.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/regression/AFTSurvivalRegression.scala
@@ -275,7 +275,7 @@ class AFTSurvivalRegression @Since("1.6.0") (@Since("1.6.0") override val uid: S
state.x.toArray.clone()
}
- bcFeaturesStd.destroy(blocking = false)
+ bcFeaturesStd.destroy()
if (handlePersistence) instances.unpersist()
val rawCoefficients = parameters.slice(2, parameters.length)
@@ -633,7 +633,7 @@ private class AFTCostFun(
case (aggregator1, aggregator2) => aggregator1.merge(aggregator2)
}, depth = aggregationDepth)
- bcParameters.destroy(blocking = false)
+ bcParameters.destroy()
(aftAggregator.loss, aftAggregator.gradient)
}
}
diff --git a/mllib/src/main/scala/org/apache/spark/ml/regression/LinearRegression.scala b/mllib/src/main/scala/org/apache/spark/ml/regression/LinearRegression.scala
index 1978287..cd0b7d3 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/regression/LinearRegression.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/regression/LinearRegression.scala
@@ -531,8 +531,8 @@ class LinearRegression @Since("1.3.0") (@Since("1.3.0") override val uid: String
throw new SparkException(msg)
}
- bcFeaturesMean.destroy(blocking = false)
- bcFeaturesStd.destroy(blocking = false)
+ bcFeaturesMean.destroy()
+ bcFeaturesStd.destroy()
val parameters = state.x.toArray.clone()
diff --git a/mllib/src/main/scala/org/apache/spark/ml/tree/impl/GradientBoostedTrees.scala b/mllib/src/main/scala/org/apache/spark/ml/tree/impl/GradientBoostedTrees.scala
index bd8c9af..c31334c 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/tree/impl/GradientBoostedTrees.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/tree/impl/GradientBoostedTrees.scala
@@ -230,7 +230,7 @@ private[spark] object GradientBoostedTrees extends Logging {
(a, b) => treesIndices.map(idx => a(idx) + b(idx)))
.map(_ / dataCount)
- broadcastTrees.destroy(blocking = false)
+ broadcastTrees.destroy()
evaluation.toArray
}
diff --git a/mllib/src/main/scala/org/apache/spark/ml/tree/impl/NodeIdCache.scala b/mllib/src/main/scala/org/apache/spark/ml/tree/impl/NodeIdCache.scala
index 5b14a63..c270e6c 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/tree/impl/NodeIdCache.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/tree/impl/NodeIdCache.scala
@@ -95,7 +95,7 @@ private[spark] class NodeIdCache(
splits: Array[Array[Split]]): Unit = {
if (prevNodeIdsForInstances != null) {
// Unpersist the previous one if one exists.
- prevNodeIdsForInstances.unpersist(false)
+ prevNodeIdsForInstances.unpersist()
}
prevNodeIdsForInstances = nodeIdsForInstances
@@ -168,11 +168,11 @@ private[spark] class NodeIdCache(
}
if (nodeIdsForInstances != null) {
// Unpersist current one if one exists.
- nodeIdsForInstances.unpersist(false)
+ nodeIdsForInstances.unpersist()
}
if (prevNodeIdsForInstances != null) {
// Unpersist the previous one if one exists.
- prevNodeIdsForInstances.unpersist(false)
+ prevNodeIdsForInstances.unpersist()
}
}
}
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/api/python/PythonMLLibAPI.scala b/mllib/src/main/scala/org/apache/spark/mllib/api/python/PythonMLLibAPI.scala
index db3f074..438a616 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/api/python/PythonMLLibAPI.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/api/python/PythonMLLibAPI.scala
@@ -98,7 +98,7 @@ private[python] class PythonMLLibAPI extends Serializable {
List(model.weights, model.intercept).map(_.asInstanceOf[Object]).asJava
}
} finally {
- data.rdd.unpersist(blocking = false)
+ data.rdd.unpersist()
}
}
@@ -336,7 +336,7 @@ private[python] class PythonMLLibAPI extends Serializable {
val model = isotonicRegressionAlg.run(input)
List[AnyRef](model.boundaryVector, model.predictionVector).asJava
} finally {
- data.rdd.unpersist(blocking = false)
+ data.rdd.unpersist()
}
}
@@ -366,7 +366,7 @@ private[python] class PythonMLLibAPI extends Serializable {
try {
kMeansAlg.run(data.rdd.persist(StorageLevel.MEMORY_AND_DISK))
} finally {
- data.rdd.unpersist(blocking = false)
+ data.rdd.unpersist()
}
}
@@ -411,7 +411,7 @@ private[python] class PythonMLLibAPI extends Serializable {
try {
new GaussianMixtureModelWrapper(gmmAlg.run(data.rdd.persist(StorageLevel.MEMORY_AND_DISK)))
} finally {
- data.rdd.unpersist(blocking = false)
+ data.rdd.unpersist()
}
}
@@ -704,7 +704,7 @@ private[python] class PythonMLLibAPI extends Serializable {
val model = word2vec.fit(dataJRDD.rdd.persist(StorageLevel.MEMORY_AND_DISK_SER))
new Word2VecModelWrapper(model)
} finally {
- dataJRDD.rdd.unpersist(blocking = false)
+ dataJRDD.rdd.unpersist()
}
}
@@ -742,7 +742,7 @@ private[python] class PythonMLLibAPI extends Serializable {
try {
DecisionTree.train(data.rdd.persist(StorageLevel.MEMORY_AND_DISK), strategy)
} finally {
- data.rdd.unpersist(blocking = false)
+ data.rdd.unpersist()
}
}
@@ -783,7 +783,7 @@ private[python] class PythonMLLibAPI extends Serializable {
RandomForest.trainRegressor(cached, strategy, numTrees, featureSubsetStrategy, intSeed)
}
} finally {
- cached.unpersist(blocking = false)
+ cached.unpersist()
}
}
@@ -814,7 +814,7 @@ private[python] class PythonMLLibAPI extends Serializable {
try {
GradientBoostedTrees.train(cached, boostingStrategy)
} finally {
- cached.unpersist(blocking = false)
+ cached.unpersist()
}
}
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/clustering/BisectingKMeans.scala b/mllib/src/main/scala/org/apache/spark/mllib/clustering/BisectingKMeans.scala
index 696dff0..b4a31d7 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/clustering/BisectingKMeans.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/clustering/BisectingKMeans.scala
@@ -217,7 +217,7 @@ class BisectingKMeans private (
newClusterCenters = newClusters.mapValues(_.center).map(identity)
}
if (preIndices != null) {
- preIndices.unpersist(false)
+ preIndices.unpersist()
}
preIndices = indices
indices = updateAssignments(assignments, divisibleIndices, newClusterCenters, dMeasure).keys
@@ -234,12 +234,12 @@ class BisectingKMeans private (
level += 1
}
if (preIndices != null) {
- preIndices.unpersist(false)
+ preIndices.unpersist()
}
if (indices != null) {
- indices.unpersist(false)
+ indices.unpersist()
}
- norms.unpersist(false)
+ norms.unpersist()
val clusters = activeClusters ++ inactiveClusters
val root = buildTree(clusters, dMeasure)
val totalCost = root.leafNodes.map(_.cost).sum
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/clustering/GaussianMixture.scala b/mllib/src/main/scala/org/apache/spark/mllib/clustering/GaussianMixture.scala
index 4d952ac..a9ed36e 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/clustering/GaussianMixture.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/clustering/GaussianMixture.scala
@@ -232,7 +232,7 @@ class GaussianMixture private (
llhp = llh // current becomes previous
llh = sums.logLikelihood // this is the freshly computed log-likelihood
iter += 1
- compute.destroy(blocking = false)
+ compute.destroy()
}
new GaussianMixtureModel(weights, gaussians)
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala b/mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala
index d967c67..d2118c2 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeans.scala
@@ -327,7 +327,7 @@ class KMeans private (
distanceMeasureInstance.centroid(sum, count)
}
- bcCenters.destroy(blocking = false)
+ bcCenters.destroy()
// Update the cluster centers and costs
converged = true
@@ -405,8 +405,8 @@ class KMeans private (
}.persist(StorageLevel.MEMORY_AND_DISK)
val sumCosts = costs.sum()
- bcNewCenters.unpersist(blocking = false)
- preCosts.unpersist(blocking = false)
+ bcNewCenters.unpersist()
+ preCosts.unpersist()
val chosen = data.zip(costs).mapPartitionsWithIndex { (index, pointCosts) =>
val rand = new XORShiftRandom(seed ^ (step << 16) ^ index)
@@ -417,8 +417,8 @@ class KMeans private (
step += 1
}
- costs.unpersist(blocking = false)
- bcNewCentersList.foreach(_.destroy(false))
+ costs.unpersist()
+ bcNewCentersList.foreach(_.destroy())
val distinctCenters = centers.map(_.vector).distinct.map(new VectorWithNorm(_))
@@ -433,7 +433,7 @@ class KMeans private (
.map(distanceMeasureInstance.findClosest(bcCenters.value, _)._1)
.countByValue()
- bcCenters.destroy(blocking = false)
+ bcCenters.destroy()
val myWeights = distinctCenters.indices.map(countMap.getOrElse(_, 0L).toDouble).toArray
LocalKMeans.kMeansPlusPlus(0, distinctCenters.toArray, myWeights, k, 30)
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeansModel.scala b/mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeansModel.scala
index b070954..32a0bff 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeansModel.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/clustering/KMeansModel.scala
@@ -103,7 +103,7 @@ class KMeansModel (@Since("1.0.0") val clusterCenters: Array[Vector],
val cost = data.map(p =>
distanceMeasureInstance.pointCost(bcCentersWithNorm.value, new VectorWithNorm(p)))
.sum()
- bcCentersWithNorm.destroy(blocking = false)
+ bcCentersWithNorm.destroy()
cost
}
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAModel.scala b/mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAModel.scala
index f915062..fc0469c 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAModel.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAModel.scala
@@ -338,7 +338,7 @@ class LocalLDAModel private[spark] (
docBound
}.sum()
- ElogbetaBc.destroy(blocking = false)
+ ElogbetaBc.destroy()
// Bound component for prob(topic-term distributions):
// E[log p(beta | eta) - log q(beta | lambda)]
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAOptimizer.scala b/mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAOptimizer.scala
index f8e5f3e..df1da62 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAOptimizer.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/clustering/LDAOptimizer.scala
@@ -505,7 +505,7 @@ final class OnlineLDAOptimizer extends LDAOptimizer with Logging {
elementWiseSum, elementWiseSum
)
- expElogbetaBc.destroy(false)
+ expElogbetaBc.destroy()
if (nonEmptyDocsN == 0) {
logWarning("No non-empty documents were submitted in the batch.")
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/feature/Word2Vec.scala b/mllib/src/main/scala/org/apache/spark/mllib/feature/Word2Vec.scala
index b8c306d..9cdade1 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/feature/Word2Vec.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/feature/Word2Vec.scala
@@ -318,9 +318,9 @@ class Word2Vec extends Serializable with Logging {
try {
doFit(dataset, sc, expTable, bcVocab, bcVocabHash)
} finally {
- expTable.destroy(blocking = false)
- bcVocab.destroy(blocking = false)
- bcVocabHash.destroy(blocking = false)
+ expTable.destroy()
+ bcVocab.destroy()
+ bcVocabHash.destroy()
}
}
@@ -450,8 +450,8 @@ class Word2Vec extends Serializable with Logging {
}
i += 1
}
- bcSyn0Global.destroy(false)
- bcSyn1Global.destroy(false)
+ bcSyn0Global.destroy()
+ bcSyn1Global.destroy()
}
newSentences.unpersist()
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala b/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala
index b2c09b4..55c2dc7 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala
@@ -179,7 +179,7 @@ class PrefixSpan private (
freqSequences.persist(data.getStorageLevel)
freqSequences.count()
}
- dataInternalRepr.unpersist(false)
+ dataInternalRepr.unpersist()
new PrefixSpanModel(freqSequences)
}
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/optimization/GradientDescent.scala b/mllib/src/main/scala/org/apache/spark/mllib/optimization/GradientDescent.scala
index 593cdd6..ffe3964 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/optimization/GradientDescent.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/optimization/GradientDescent.scala
@@ -246,7 +246,7 @@ object GradientDescent extends Logging {
// c: (grad, loss, count)
(c1._1 += c2._1, c1._2 + c2._2, c1._3 + c2._3)
})
- bcWeights.destroy(blocking = false)
+ bcWeights.destroy()
if (miniBatchSize > 0) {
/**
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/optimization/LBFGS.scala b/mllib/src/main/scala/org/apache/spark/mllib/optimization/LBFGS.scala
index 21ec287..65c1159 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/optimization/LBFGS.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/optimization/LBFGS.scala
@@ -261,7 +261,7 @@ object LBFGS extends Logging {
val (gradientSum, lossSum) = data.treeAggregate((zeroSparseVector, 0.0))(seqOp, combOp)
// broadcasted model is not needed anymore
- bcW.destroy(blocking = false)
+ bcW.destroy()
/**
* regVal is sum of weight squares if it's L2 updater;
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/regression/GeneralizedLinearAlgorithm.scala b/mllib/src/main/scala/org/apache/spark/mllib/regression/GeneralizedLinearAlgorithm.scala
index 4d5aaf1..ef3ff2b 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/regression/GeneralizedLinearAlgorithm.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/regression/GeneralizedLinearAlgorithm.scala
@@ -366,7 +366,7 @@ abstract class GeneralizedLinearAlgorithm[M <: GeneralizedLinearModel]
// Unpersist cached data
if (data.getStorageLevel != StorageLevel.NONE) {
- data.unpersist(false)
+ data.unpersist()
}
createModel(weights, intercept)
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/tree/model/treeEnsembleModels.scala b/mllib/src/main/scala/org/apache/spark/mllib/tree/model/treeEnsembleModels.scala
index b1e8265..fc1d412 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/tree/model/treeEnsembleModels.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/tree/model/treeEnsembleModels.scala
@@ -167,7 +167,7 @@ class GradientBoostedTreesModel @Since("1.2.0") (
(a, b) => treesIndices.map(idx => a(idx) + b(idx)))
.map(_ / dataCount)
- broadcastTrees.destroy(blocking = false)
+ broadcastTrees.destroy()
evaluation.toArray
}
diff --git a/mllib/src/test/scala/org/apache/spark/ml/classification/LinearSVCSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/classification/LinearSVCSuite.scala
index c05c896..cb9b8f9 100644
--- a/mllib/src/test/scala/org/apache/spark/ml/classification/LinearSVCSuite.scala
+++ b/mllib/src/test/scala/org/apache/spark/ml/classification/LinearSVCSuite.scala
@@ -181,8 +181,8 @@ class LinearSVCSuite extends MLTest with DefaultReadWriteTest {
}
assert(thrown.getMessage.contains("coefficients only supports dense"))
- bcCoefficients.destroy(blocking = false)
- bcFeaturesStd.destroy(blocking = false)
+ bcCoefficients.destroy()
+ bcFeaturesStd.destroy()
}
test("linearSVC with sample weights") {
diff --git a/mllib/src/test/scala/org/apache/spark/ml/classification/LogisticRegressionSuite.scala b/mllib/src/test/scala/org/apache/spark/ml/classification/LogisticRegressionSuite.scala
index 84c10e2..2499892 100644
--- a/mllib/src/test/scala/org/apache/spark/ml/classification/LogisticRegressionSuite.scala
+++ b/mllib/src/test/scala/org/apache/spark/ml/classification/LogisticRegressionSuite.scala
@@ -549,9 +549,9 @@ class LogisticRegressionSuite extends MLTest with DefaultReadWriteTest {
}
}
assert(thrown.getMessage.contains("coefficients only supports dense"))
- bcCoefficientsBinary.destroy(blocking = false)
- bcFeaturesStd.destroy(blocking = false)
- bcCoefficientsMulti.destroy(blocking = false)
+ bcCoefficientsBinary.destroy()
+ bcFeaturesStd.destroy()
+ bcCoefficientsMulti.destroy()
}
test("overflow prediction for multiclass") {
diff --git a/python/pyspark/broadcast.py b/python/pyspark/broadcast.py
index 69447a6..43a5ead 100644
--- a/python/pyspark/broadcast.py
+++ b/python/pyspark/broadcast.py
@@ -159,16 +159,19 @@ class Broadcast(object):
raise Exception("Broadcast can only be unpersisted in driver")
self._jbroadcast.unpersist(blocking)
- def destroy(self):
+ def destroy(self, blocking=False):
"""
Destroy all data and metadata related to this broadcast variable.
Use this with caution; once a broadcast variable has been destroyed,
- it cannot be used again. This method blocks until destroy has
- completed.
+ it cannot be used again.
+
+ .. versionchanged:: 3.0.0
+ Added optional argument `blocking` to specify whether to block until all
+ blocks are deleted.
"""
if self._jbroadcast is None:
raise Exception("Broadcast can only be destroyed in driver")
- self._jbroadcast.destroy()
+ self._jbroadcast.destroy(blocking)
os.unlink(self._path)
def __reduce__(self):
diff --git a/python/pyspark/rdd.py b/python/pyspark/rdd.py
index 751df44..ef382f3 100644
--- a/python/pyspark/rdd.py
+++ b/python/pyspark/rdd.py
@@ -244,13 +244,17 @@ class RDD(object):
self._jrdd.persist(javaStorageLevel)
return self
- def unpersist(self):
+ def unpersist(self, blocking=False):
"""
Mark the RDD as non-persistent, and remove all blocks for it from
memory and disk.
+
+ .. versionchanged:: 3.0.0
+ Added optional argument `blocking` to specify whether to block until all
+ blocks are deleted.
"""
self.is_cached = False
- self._jrdd.unpersist()
+ self._jrdd.unpersist(blocking)
return self
def checkpoint(self):
diff --git a/python/pyspark/tests/test_rdd.py b/python/pyspark/tests/test_rdd.py
index b2a544b..bac2f78 100644
--- a/python/pyspark/tests/test_rdd.py
+++ b/python/pyspark/tests/test_rdd.py
@@ -324,7 +324,7 @@ class RDDTests(ReusedPySparkTestCase):
bdata.unpersist()
m = self.sc.parallelize(range(1), 1).map(lambda x: len(bdata.value)).sum()
self.assertEqual(N, m)
- bdata.destroy()
+ bdata.destroy(blocking=True)
try:
self.sc.parallelize(range(1), 1).map(lambda x: len(bdata.value)).sum()
except Exception as e:
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
index b7fc957..16b14a6 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameFunctionsSuite.scala
@@ -1395,7 +1395,7 @@ class DataFrameFunctionsSuite extends QueryTest with SharedSQLContext {
}
// Test with local relation, the Project will be evaluated without codegen
- df.unpersist()
+ df.unpersist(blocking = true)
nullTest()
// Test with cached relation, the Project will be evaluated with codegen
df.cache()
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DatasetCacheSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DatasetCacheSuite.scala
index 1ff9f6d..b3b2cdc 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DatasetCacheSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DatasetCacheSuite.scala
@@ -49,7 +49,7 @@ class DatasetCacheSuite extends QueryTest with SharedSQLContext with TimeLimits
assert(ds1.storageLevel == StorageLevel.MEMORY_AND_DISK)
assert(ds2.storageLevel == StorageLevel.MEMORY_AND_DISK)
// unpersist
- ds1.unpersist()
+ ds1.unpersist(blocking = true)
assert(ds1.storageLevel == StorageLevel.NONE)
// non-default storage level
ds1.persist(StorageLevel.MEMORY_ONLY_2)
@@ -71,7 +71,7 @@ class DatasetCacheSuite extends QueryTest with SharedSQLContext with TimeLimits
cached,
2, 3, 4)
// Drop the cache.
- cached.unpersist()
+ cached.unpersist(blocking = true)
assert(cached.storageLevel == StorageLevel.NONE, "The Dataset should not be cached.")
}
@@ -88,9 +88,9 @@ class DatasetCacheSuite extends QueryTest with SharedSQLContext with TimeLimits
checkDataset(joined, ("2", 2))
assertCached(joined, 2)
- ds1.unpersist()
+ ds1.unpersist(blocking = true)
assert(ds1.storageLevel == StorageLevel.NONE, "The Dataset ds1 should not be cached.")
- ds2.unpersist()
+ ds2.unpersist(blocking = true)
assert(ds2.storageLevel == StorageLevel.NONE, "The Dataset ds2 should not be cached.")
}
@@ -105,9 +105,9 @@ class DatasetCacheSuite extends QueryTest with SharedSQLContext with TimeLimits
("b", 3))
assertCached(agged.filter(_._1 == "b"))
- ds.unpersist()
+ ds.unpersist(blocking = true)
assert(ds.storageLevel == StorageLevel.NONE, "The Dataset ds should not be cached.")
- agged.unpersist()
+ agged.unpersist(blocking = true)
assert(agged.storageLevel == StorageLevel.NONE, "The Dataset agged should not be cached.")
}
@@ -122,7 +122,7 @@ class DatasetCacheSuite extends QueryTest with SharedSQLContext with TimeLimits
df.count()
assertCached(df2)
- df.unpersist()
+ df.unpersist(blocking = true)
assert(df.storageLevel == StorageLevel.NONE)
}
@@ -140,7 +140,7 @@ class DatasetCacheSuite extends QueryTest with SharedSQLContext with TimeLimits
df2.collect()
}
- df.unpersist()
+ df.unpersist(blocking = true)
assert(df.storageLevel == StorageLevel.NONE)
}
@@ -166,7 +166,7 @@ class DatasetCacheSuite extends QueryTest with SharedSQLContext with TimeLimits
df.count()
df3.cache()
- df.unpersist()
+ df.unpersist(blocking = true)
// df un-cached; df2 and df3's cache plan re-compiled
assert(df.storageLevel == StorageLevel.NONE)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala
index 861aa17..1c0d603 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/InMemoryColumnarQuerySuite.scala
@@ -429,7 +429,7 @@ class InMemoryColumnarQuerySuite extends QueryTest with SharedSQLContext {
val df2 = Seq(("a", 1), ("b", 2), ("c", 3)).toDF("item", "id")
val df3 = df1.join(df2, Seq("item")).select($"id", $"group".as("item")).distinct()
- df3.unpersist()
+ df3.unpersist(blocking = true)
val agg_without_cache = df3.groupBy($"item").count()
df3.cache()
@@ -445,7 +445,7 @@ class InMemoryColumnarQuerySuite extends QueryTest with SharedSQLContext {
// with a non-empty list
assert(df.filter($"id".isin(2)).count() == 1)
assert(df.filter($"id".isin(2, 3)).count() == 2)
- df.unpersist()
+ df.unpersist(blocking = true)
val dfNulls = spark.range(10).selectExpr("null as id").cache()
// with null as value for the attribute
assert(dfNulls.filter($"id".isin()).count() == 0)
@@ -466,7 +466,7 @@ class InMemoryColumnarQuerySuite extends QueryTest with SharedSQLContext {
testWithWholeStageCodegenOnAndOff("SPARK-22348: table cache " +
"should do partition batch pruning") { codegenEnabled =>
val df1 = Seq((1, 1), (1, 1), (2, 2)).toDF("x", "y")
- df1.unpersist()
+ df1.unpersist(blocking = true)
df1.cache()
// Push predicate to the cached table.
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/CachedTableSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/CachedTableSuite.scala
index b492f39..9483a9c 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/CachedTableSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/CachedTableSuite.scala
@@ -123,7 +123,7 @@ class CachedTableSuite extends QueryTest with SQLTestUtils with TestHiveSingleto
// no error will be reported in the following three ways to uncache a table.
spark.catalog.uncacheTable(tableName)
sql("UNCACHE TABLE newTable")
- sparkSession.table(tableName).unpersist()
+ sparkSession.table(tableName).unpersist(blocking = true)
}
}
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala b/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala
index 3524337..41374b5 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala
@@ -455,7 +455,7 @@ abstract class DStream[T: ClassTag] (
if (unpersistData) {
logDebug(s"Unpersisting old RDDs: ${oldRDDs.values.map(_.id).mkString(", ")}")
oldRDDs.values.foreach { rdd =>
- rdd.unpersist(false)
+ rdd.unpersist()
// Explicitly remove blocks of BlockRDD
rdd match {
case b: BlockRDD[_] =>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org