[SPARK-44398][CONNECT] Scala foreachBatch API #41969
Closed
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
|
cc: @bogao007 |
bogao007
approved these changes
Jul 12, 2023
| * Handles setting up Scala remote session and other Spark Connect environment and then | ||
| * runs the provided foreachBatch function `fn`. | ||
| * | ||
| * HACK ALERT: This version does not atually set up Spark connect. Directly passes the DataFrame, |
There was a problem hiding this comment.
I have 2 major questions regarding this:
- Is the missing part about setting up a Spark Connect session and converting the legacy DataFrame to Spark Connect DataFrame and being executed inside the Spark Connect session? Do we have any Scala example on setting up Spark Connect session on server side and use it?
- When is
getDataFrameOrThrow()being called? Is it only needed for Python or do we need to get the DataFrame by ID inside the Spark Connect session for Scala?
There was a problem hiding this comment.
- Yes, it is about setting up spark remote session. I don't think there are examples of doing that in Scala.
- Not sure about the second one. Usually
df.sparkSessiongives the access to session.
...r/connect/client/jvm/src/test/scala/org/apache/spark/sql/streaming/StreamingQuerySuite.scala
Outdated
Show resolved
Hide resolved
|
@zhenlineo could you review this? I am going to get the test working right. |
zhenlineo
approved these changes
Jul 13, 2023
xinrong-meng
approved these changes
Jul 13, 2023
|
Merged to master, thanks! |
ragnarok56
pushed a commit
to ragnarok56/spark
that referenced
this pull request
Mar 2, 2024
This implements Scala foreachBatch(). The implementation basic and needs some more enhancements. The server side will be shared by Python implementation as well. One notable hack in this PR is that it runs user's `foreachBatch()` with regular(legacy) DataFrame, rather than setting up remote Spark connect session and connect DataFrame. ### Why are the changes needed? Adds foreachBatch() support in Scala Spark Connect. ### Does this PR introduce _any_ user-facing change? Yes. Adds foreachBatch() API ### How was this patch tested? - A simple unit test. Closes apache#41969 from rangadi/feb-scala. Authored-by: Raghu Angadi <raghu.angadi@databricks.com> Signed-off-by: Xinrong Meng <xinrong@apache.org>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This implements Scala foreachBatch(). The implementation basic and needs some more enhancements. The server side will be shared by Python implementation as well.
One notable hack in this PR is that it runs user's
foreachBatch()with regular(legacy) DataFrame, rather than setting up remote Spark connect session and connect DataFrame.Why are the changes needed?
Adds foreachBatch() support in Scala Spark Connect.
Does this PR introduce any user-facing change?
Yes. Adds foreachBatch() API
How was this patch tested?