You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "HyukjinKwon (via GitHub)" <gi...@apache.org> on 2024/05/21 10:48:51 UTC
[PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
HyukjinKwon opened a new pull request, #46683:
URL: https://github.com/apache/spark/pull/46683
### What changes were proposed in this pull request?
This PR adds `Dataset.checkpoint` and `Dataset.localCheckpoint` into Scala Spark Connect client. Python API was implemented at https://github.com/apache/spark/pull/46570
### Why are the changes needed?
For API parity.
### Does this PR introduce _any_ user-facing change?
Yes, it adds `Dataset.checkpoint` and `Dataset.localCheckpoint` into Scala Spark Connect client.
### How was this patch tested?
Unittests added.
### Was this patch authored or co-authored using generative AI tooling?
No.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1608572841
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -3468,6 +3554,26 @@ class Dataset[T] private[sql] (
}
}
+ // Visible for testing
+ private[sql] var cachedRemoteRelationID: Option[String] = None
+
+ override def finalize(): Unit = {
Review Comment:
I also think this is an issue for python.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1609144681
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -3468,6 +3554,26 @@ class Dataset[T] private[sql] (
}
}
+ // Visible for testing
+ private[sql] var cachedRemoteRelationID: Option[String] = None
+
+ override def finalize(): Unit = {
Review Comment:
For Python, https://github.com/apache/spark/pull/46694
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1609225579
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -73,6 +73,11 @@ class SparkSession private[sql] (
with Logging {
private[this] val allocator = new RootAllocator()
+ private var shouldStopCleaner = false
+ private[sql] lazy val cleaner = {
+ shouldStopCleaner = true
+ new ContextCleaner(this)
Review Comment:
And seems like Java protobuf gen does not allow it:
> Don’t go looking for facilities similar to class inheritance, though – protocol buffers don’t do that.
https://protobuf.dev/getting-started/javatutorial/#defining-your-protocol-format
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #46683:
URL: https://github.com/apache/spark/pull/46683#issuecomment-2126156236
I believe I addressed all major comments here. I need to make one more followup to fix both Scala/Python. Let me merge this one. and address others in that PR if there are more to fix.
Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon closed pull request #46683: [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client
URL: https://github.com/apache/spark/pull/46683
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1609227214
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -73,6 +73,11 @@ class SparkSession private[sql] (
with Logging {
private[this] val allocator = new RootAllocator()
+ private var shouldStopCleaner = false
+ private[sql] lazy val cleaner = {
+ shouldStopCleaner = true
+ new ContextCleaner(this)
Review Comment:
Let me make it simpler at least.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1609193650
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/internal/ContextCleaner.scala:
##########
@@ -0,0 +1,176 @@
+/*
Review Comment:
I referred https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/ContextCleaner.scala
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1608570707
##########
connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/ClientE2ETestSuite.scala:
##########
@@ -1554,6 +1555,37 @@ class ClientE2ETestSuite extends RemoteSparkSession with SQLHelper with PrivateM
val metrics = SparkThreadUtils.awaitResult(future, 2.seconds)
assert(metrics === Map("min(id)" -> 0, "avg(id)" -> 49, "max(id)" -> 98))
}
+
+ test("checkpoint") {
+ val df = spark.range(100).localCheckpoint()
+ testCapturedStdOut(df.explain(), "ExistingRDD")
+ }
+
+ test("checkpoint gc") {
+ var df1 = spark.range(100).localCheckpoint(eager = true)
+ val encoder = df1.agnosticEncoder
+ val dfId = df1.cachedRemoteRelationID.get
+
+ // GC triggers remove the cached remote relation
+ df1 = null
+ System.gc()
+
+ // Make sure the cleanup happens in the server side.
+ Thread.sleep(3000L)
Review Comment:
I can guarantee you that this will be flaky. IMO we assume that GC and reference queues work. The only thing we need to test if the action called upon GC is working.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1609058630
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -3468,6 +3554,26 @@ class Dataset[T] private[sql] (
}
}
+ // Visible for testing
+ private[sql] var cachedRemoteRelationID: Option[String] = None
+
+ override def finalize(): Unit = {
Review Comment:
For Python actually there's no alternative, and `__del__` isn't actually deprecated in Python. I read why it's deprecated in JDK (e.g., no guarantee on the order of `fianlize` invoke, etc.), and it won't affect this specific usecase but yeah let me switch it to `java.lang.ref.Cleaner`.
> This assumes that the original dataframe is always in scope, however if that is being garbage collected any derived dataframe breaks. I think we should pin it to the original CachedRemoteRelation instead, that will be re-used in derived dataframes.
Yeah, this is a good point. Let me fix it up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1609059212
##########
connector/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClient.scala:
##########
@@ -39,7 +39,7 @@ import org.apache.spark.sql.connect.common.config.ConnectCommon
*/
private[sql] class SparkConnectClient(
private[sql] val configuration: SparkConnectClient.Configuration,
- private val channel: ManagedChannel) {
+ private[sql] val channel: ManagedChannel) {
Review Comment:
This is used to check if the client is shut down before trying to remove the remote cached relation at `finalize`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1608571929
##########
connector/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClient.scala:
##########
@@ -39,7 +39,7 @@ import org.apache.spark.sql.connect.common.config.ConnectCommon
*/
private[sql] class SparkConnectClient(
private[sql] val configuration: SparkConnectClient.Configuration,
- private val channel: ManagedChannel) {
+ private[sql] val channel: ManagedChannel) {
Review Comment:
Why is this needed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1608527529
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -3468,6 +3554,26 @@ class Dataset[T] private[sql] (
}
}
+ // Visible for testing
+ private[sql] var cachedRemoteRelationID: Option[String] = None
+
+ override def finalize(): Unit = {
Review Comment:
Finalize has been deprecated in java 9 for good reasons. Let's use a `java.lang.ref.Cleaner` instead.
However high level I have a different concern. This assumes that the original dataframe is always in scope, however if that is being garbage collected any derived dataframe breaks. I think we should pin it to the original CachedRemoteRelation instead, that will be re-used in derived dataframes. The upside of this is that you don't have to introduce a new field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "hvanhovell (via GitHub)" <gi...@apache.org>.
hvanhovell commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1609208957
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -73,6 +73,11 @@ class SparkSession private[sql] (
with Logging {
private[this] val allocator = new RootAllocator()
+ private var shouldStopCleaner = false
+ private[sql] lazy val cleaner = {
+ shouldStopCleaner = true
+ new ContextCleaner(this)
Review Comment:
Why not use the JVM one?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48370][CONNECT] Checkpoint and localCheckpoint in Scala Spark Connect client [spark]
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #46683:
URL: https://github.com/apache/spark/pull/46683#discussion_r1609223886
##########
connector/connect/client/jvm/src/main/scala/org/apache/spark/sql/SparkSession.scala:
##########
@@ -73,6 +73,11 @@ class SparkSession private[sql] (
with Logging {
private[this] val allocator = new RootAllocator()
+ private var shouldStopCleaner = false
+ private[sql] lazy val cleaner = {
+ shouldStopCleaner = true
+ new ContextCleaner(this)
Review Comment:
The problem is that the interface forces to fix generated code `org.apache.spark.connect.proto.CachedRemoteRelation` to inherit `AutoCloseable`, or I have to create a wrapper (see https://docs.oracle.com/javase/9/docs/api/java/lang/ref/Cleaner.html)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org