You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "WeichenXu123 (via GitHub)" <gi...@apache.org> on 2023/05/29 05:37:19 UTC
[GitHub] [spark] WeichenXu123 opened a new pull request, #41357: [SPARK-43790] [PYSPARK][CONNECT][ML] Add `copyFromLocalToFS` API
WeichenXu123 opened a new pull request, #41357:
URL: https://github.com/apache/spark/pull/41357
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
Add `copyFromLocalToFS` API.
The API copies a provided local file to provided destination path on hadoop FS.
The implementation of the API reuses `AddArtifact` proto message, and it fill the Artifact path as `forward_to_fs/{destination_path}` , and in spark connect server backend, when handling artifact requests, if receiving an artifact request with path starting with `forward_to_fs`, it saves the artifact data to the provided `destination_path` on HadoopFS.
### Why are the changes needed?
In new distributed spark ML module (designed to support spark connect and support local inference)
We need to save ML model to hadoop file system as its raw file format and saved as provided file name, the reason is:
We often submit a spark application to spark cluster for running the training model job, we need to save trained model to hadoop file system before the spark application completes.
But we want to support local model inference, that means if we save the model by current spark DataFrame writer (e.g. parquet format), when loading model we have to rely on the spark service. But we hope we can load model without spark service. So we want the model being saved as the original binary format that our ML code can handle.
### Does this PR introduce _any_ user-facing change?
Yes
Add `copyFromLocalToFS` API:
```
def copyFromLocalToFS(self, local_path: str, dest_path: str) -> None:
"""
copy file from local to FS.
Parameters
----------
local_path: str
Path to a local file (Directory is not supported yet).
The path can be either absolute path or relative path.
dest_path: str
The FS path to the destination the file will be copied to.
.. versionadded:: 3.5.0
"""
```
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
Unit tests.
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1220721480
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala:
##########
@@ -157,10 +159,46 @@ class SparkConnectArtifactManager private[connect] {
}
}
}
+
+ private[connect] def uploadArtifactToFs(
+ sessionHolder: SessionHolder,
+ remoteRelativePath: Path,
+ serverLocalStagingPath: Path): Unit = {
+ val hadoopConf = sessionHolder.session.sparkContext.hadoopConfiguration
+ assert(
+ remoteRelativePath.startsWith(
+ SparkConnectArtifactManager.forwardToFSPrefix + File.separator))
+ val destFSPath = new FSPath(
+ Paths
+ .get("/")
+ .resolve(remoteRelativePath.subpath(1, remoteRelativePath.getNameCount))
+ .toString)
+ val localPath = serverLocalStagingPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ if (fs.isInstanceOf[LocalFileSystem]) {
+ val allowDestLocalConf = sessionHolder.session.sparkContext.conf
+ .get("spark.fs.copyFromLocalToFs.allowDestLocal", "false")
Review Comment:
I think we should better add this conf to `org.apache.spark.sql.connect.config.Connect`, and explicitly document that it's for testing purpose.
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212596759
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala:
##########
@@ -157,10 +159,34 @@ class SparkConnectArtifactManager private[connect] {
}
}
}
+
+ private[connect] def uploadArtifactToFs(
+ sessionHolder: SessionHolder,
+ remoteRelativePath: Path,
+ serverLocalStagingPath: Path): Unit = {
+ val hadoopConf = sessionHolder.session.sparkContext.hadoopConfiguration
+ assert(remoteRelativePath.startsWith(
+ SparkConnectArtifactManager.forwardToFSPrefix + File.separator)
+ )
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(
+ remoteRelativePath.subpath(1, remoteRelativePath.getNameCount)).toString
+ )
+ val localPath = serverLocalStagingPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
Review Comment:
The code here saves file to cloud storage file system.
--
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
[GitHub] [spark] vicennial commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1210431327
##########
python/pyspark/sql/tests/connect/client/test_artifact.py:
##########
@@ -271,6 +271,21 @@ def func(x):
self.spark.addArtifacts(f"{archive_path}.zip#my_files", archive=True)
self.assertEqual(self.spark.range(1).select(func("id")).first()[0], "hello world!")
+ def test_copy_from_local_to_FS(self):
Review Comment:
Please add a server/scala-side test as well in `ArtifactManagerSuite`
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
Review Comment:
Could we move this logic inside of `SparkConnectArtifactManager#AddArtifact` too unless there's a specific reason it needs to be here?
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1211157822
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
+ try {
+ val hadoopConf = holder.session.sparkContext.hadoopConfiguration
+ assert(artifact.path.startsWith(forwardToFSPrefix))
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(artifact.path.subpath(1, artifact.path.getNameCount)).toString
+ )
+ val localPath = artifact.stagedPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
+ )
+ } catch {
+ case e: java.io.IOException =>
+ // TODO: report failure in response message
Review Comment:
`SparkConnectAddArtifactsHandler` already defined `onError` handler (note this class inherits `StreamObserver`) , see
https://github.com/apache/spark/blob/e8f87413bdd905e5e208ac8484b3b11de1623fc2/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala#L83
We might need to do some refactor to make it use `handleError` , @allisonwang-db would you mind making the change in follow-up PR ?
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212562805
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -100,7 +100,15 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
// We do not store artifacts that fail the CRC. The failure is reported in the artifact
// summary and it is up to the client to decide whether to retry sending the artifact.
if (artifact.getCrcStatus.contains(true)) {
- addStagedArtifactToArtifactManager(artifact)
+ if (artifact.path.startsWith(
Review Comment:
The goal for artifact with `forward_to_fs/{dest_fs_path}` path is just to upload file to cloud storage. Do we still need add the artifact file to artifact manager ?
If you want , I can add it, it is simple.
--
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
[GitHub] [spark] WeichenXu123 closed pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 closed pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
URL: https://github.com/apache/spark/pull/41357
--
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
[GitHub] [spark] allisonwang-db commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212021765
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
+ try {
+ val hadoopConf = holder.session.sparkContext.hadoopConfiguration
+ assert(artifact.path.startsWith(forwardToFSPrefix))
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(artifact.path.subpath(1, artifact.path.getNameCount)).toString
+ )
+ val localPath = artifact.stagedPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
+ )
+ } catch {
+ case e: java.io.IOException =>
+ // TODO: report failure in response message
Review Comment:
@MaxGekk will be working on a new framework for error handling. @WeichenXu123 You can migrate to use the new framework when it's completed.
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212596759
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala:
##########
@@ -157,10 +159,34 @@ class SparkConnectArtifactManager private[connect] {
}
}
}
+
+ private[connect] def uploadArtifactToFs(
+ sessionHolder: SessionHolder,
+ remoteRelativePath: Path,
+ serverLocalStagingPath: Path): Unit = {
+ val hadoopConf = sessionHolder.session.sparkContext.hadoopConfiguration
+ assert(remoteRelativePath.startsWith(
+ SparkConnectArtifactManager.forwardToFSPrefix + File.separator)
+ )
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(
+ remoteRelativePath.subpath(1, remoteRelativePath.getNameCount)).toString
+ )
+ val localPath = serverLocalStagingPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
Review Comment:
Save file to cloud storage file system.
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212526198
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -100,7 +100,15 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
// We do not store artifacts that fail the CRC. The failure is reported in the artifact
// summary and it is up to the client to decide whether to retry sending the artifact.
if (artifact.getCrcStatus.contains(true)) {
- addStagedArtifactToArtifactManager(artifact)
+ if (artifact.path.startsWith(
Review Comment:
Hm, shouldn't we do this at `addStagedArtifactToArtifactManager`?
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1209592895
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
+ try {
+ val hadoopConf = holder.session.sparkContext.hadoopConfiguration
+ assert(artifact.path.startsWith(forwardToFSPrefix))
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(artifact.path.subpath(1, artifact.path.getNameCount)).toString
+ )
+ val localPath = artifact.stagedPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
+ )
+ } catch {
+ case e: java.io.IOException =>
+ // TODO: report failure in response message
Review Comment:
I believe @allisonwang-db is working on that.
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1220726011
##########
python/pyspark/sql/connect/session.py:
##########
@@ -625,6 +625,26 @@ def addArtifacts(self, *path: str, pyfile: bool = False, archive: bool = False)
addArtifact = addArtifacts
+ def copyFromLocalToFs(self, local_path: str, dest_path: str) -> None:
Review Comment:
Can we add `.. versionadded 3.5.0` directive?
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1210942943
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
Review Comment:
No problem. :)
##########
python/pyspark/sql/tests/connect/client/test_artifact.py:
##########
@@ -271,6 +271,21 @@ def func(x):
self.spark.addArtifacts(f"{archive_path}.zip#my_files", archive=True)
self.assertEqual(self.spark.range(1).select(func("id")).first()[0], "hello world!")
+ def test_copy_from_local_to_FS(self):
Review Comment:
Sure.
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1211149046
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
+ try {
+ val hadoopConf = holder.session.sparkContext.hadoopConfiguration
+ assert(artifact.path.startsWith(forwardToFSPrefix))
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(artifact.path.subpath(1, artifact.path.getNameCount)).toString
+ )
+ val localPath = artifact.stagedPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
+ )
+ } catch {
+ case e: java.io.IOException =>
+ // TODO: report failure in response message
Review Comment:
@allisonwang-db
`handleError` is defined in `SparkConnectService` and it is private, but my code is in `SparkConnectAddArtifactsHandler` class, how can I call `handleError` from `SparkConnectAddArtifactsHandler` code ? Could you make `handleError` to be a utility function that is accessible form other classes ?
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212603254
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala:
##########
@@ -157,10 +159,34 @@ class SparkConnectArtifactManager private[connect] {
}
}
}
+
+ private[connect] def uploadArtifactToFs(
+ sessionHolder: SessionHolder,
+ remoteRelativePath: Path,
+ serverLocalStagingPath: Path): Unit = {
+ val hadoopConf = sessionHolder.session.sparkContext.hadoopConfiguration
+ assert(remoteRelativePath.startsWith(
+ SparkConnectArtifactManager.forwardToFSPrefix + File.separator)
Review Comment:
🆗
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYSPARK][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1209105558
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
+ try {
+ val hadoopConf = holder.session.sparkContext.hadoopConfiguration
+ assert(artifact.path.startsWith(forwardToFSPrefix))
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(artifact.path.subpath(1, artifact.path.getNameCount)).toString
+ )
+ val localPath = artifact.stagedPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
+ )
+ } catch {
+ case e: java.io.IOException =>
+ // TODO: report failure in response message
Review Comment:
Discussion: Shall we add a field in ArtifactResponse message for holding failure information here ?
--
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
[GitHub] [spark] HyukjinKwon commented on pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on PR #41357:
URL: https://github.com/apache/spark/pull/41357#issuecomment-1571388121
Looks fine to me but I would leave it to @hvanhovell and @vicennial to check and merge.
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212522100
##########
python/pyspark/sql/connect/client/artifact.py:
##########
@@ -187,6 +188,19 @@ def _parse_artifacts(self, path_or_uri: str, pyfile: bool, archive: bool) -> Lis
return [artifact]
raise RuntimeError(f"Unsupported scheme: {parsed.scheme}")
+ def _parse_forward_to_FS_artifacts(self, local_path: str, dest_path: str) -> List[Artifact]:
Review Comment:
```suggestion
def _parse_forward_to_fs_artifacts(self, local_path: str, dest_path: str) -> List[Artifact]:
```
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212554762
##########
python/pyspark/sql/connect/client/artifact.py:
##########
@@ -187,6 +188,19 @@ def _parse_artifacts(self, path_or_uri: str, pyfile: bool, archive: bool) -> Lis
return [artifact]
raise RuntimeError(f"Unsupported scheme: {parsed.scheme}")
+ def _parse_forward_to_FS_artifacts(self, local_path: str, dest_path: str) -> List[Artifact]:
Review Comment:
I don't think so. The arguments are different. For `_parse_forward_to_FS_artifacts`, we have `local_path` and `dest_path` and the 2 paths are different.
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1220735232
##########
python/pyspark/sql/connect/session.py:
##########
@@ -637,6 +638,34 @@ def addArtifacts(self, *path: str, pyfile: bool = False, archive: bool = False)
addArtifact = addArtifacts
+ def copyFromLocalToFs(self, local_path: str, dest_path: str) -> None:
+ """
+ Copy file from local to cloud storage file system.
+ If the file already exits in destination path, old file is overwritten.
+
+ Parameters
+ ----------
+ local_path: str
+ Path to a local file (Directory is not supported yet).
+ The path can be either an absolute path or a relative path.
+
+ dest_path: str
+ The cloud storage path to the destination the file will
+ be copied to.
+ The path must be an an absolute path.
+
+ Notes
+ -----
+ This API is a developer API.
+ """
+ if urllib.parse.urlparse(dest_path).scheme:
+ raise ValueError(
+ "`spark_session.copyFromLocalToFs` API only allows `dest_path` to be a path "
+ "without scheme, and spark driver uses the default scheme to "
+ "determine the destination file system."
Review Comment:
Why can't we support the fully qualified URI with the scheme? We just add a prefix and truncate it later in the server.
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1209592895
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
+ try {
+ val hadoopConf = holder.session.sparkContext.hadoopConfiguration
+ assert(artifact.path.startsWith(forwardToFSPrefix))
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(artifact.path.subpath(1, artifact.path.getNameCount)).toString
+ )
+ val localPath = artifact.stagedPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
+ )
+ } catch {
+ case e: java.io.IOException =>
+ // TODO: report failure in response message
Review Comment:
I believe @allisonwang-db is working on that (adding a way of passing an error throw RPC). She'd know some context here.
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212521469
##########
python/pyspark/sql/connect/session.py:
##########
@@ -625,6 +625,26 @@ def addArtifacts(self, *path: str, pyfile: bool = False, archive: bool = False)
addArtifact = addArtifacts
+ def copyFromLocalToFs(self, local_path: str, dest_path: str) -> None:
Review Comment:
Can be move this to ML package? Should also mark it as a developer API in `Notes` section.
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212561246
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala:
##########
@@ -157,10 +159,34 @@ class SparkConnectArtifactManager private[connect] {
}
}
}
+
+ private[connect] def uploadArtifactToFs(
+ sessionHolder: SessionHolder,
+ remoteRelativePath: Path,
+ serverLocalStagingPath: Path): Unit = {
+ val hadoopConf = sessionHolder.session.sparkContext.hadoopConfiguration
+ assert(remoteRelativePath.startsWith(
+ SparkConnectArtifactManager.forwardToFSPrefix + File.separator)
Review Comment:
In the design suggested by @vicennial , We defined the artifact path to be `forward_to_fs/{dest_fs_path}`, and `dest_fs_path` is absolute path on cloud storage. I think it is fine and it works in all cases.
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1221167380
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala:
##########
@@ -157,10 +159,46 @@ class SparkConnectArtifactManager private[connect] {
}
}
}
+
+ private[connect] def uploadArtifactToFs(
+ sessionHolder: SessionHolder,
+ remoteRelativePath: Path,
+ serverLocalStagingPath: Path): Unit = {
+ val hadoopConf = sessionHolder.session.sparkContext.hadoopConfiguration
+ assert(
+ remoteRelativePath.startsWith(
+ SparkConnectArtifactManager.forwardToFSPrefix + File.separator))
+ val destFSPath = new FSPath(
+ Paths
+ .get("/")
+ .resolve(remoteRelativePath.subpath(1, remoteRelativePath.getNameCount))
+ .toString)
+ val localPath = serverLocalStagingPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ if (fs.isInstanceOf[LocalFileSystem]) {
+ val allowDestLocalConf = sessionHolder.session.sparkContext.conf
+ .get("spark.fs.copyFromLocalToFs.allowDestLocal", "false")
Review Comment:
Sounds good.
--
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
[GitHub] [spark] vicennial commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "vicennial (via GitHub)" <gi...@apache.org>.
vicennial commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1217977040
##########
python/pyspark/sql/connect/client/artifact.py:
##########
@@ -187,6 +188,19 @@ def _parse_artifacts(self, path_or_uri: str, pyfile: bool, archive: bool) -> Lis
return [artifact]
raise RuntimeError(f"Unsupported scheme: {parsed.scheme}")
+ def _parse_forward_to_fs_artifacts(self, local_path: str, dest_path: str) -> List[Artifact]:
+ local_path = Path(local_path).absolute()
+ # TODO: Support directory path.
+ assert local_path.is_file(), "local path must be a file path."
+ storage = LocalFile(str(local_path))
+
+ assert Path(dest_path).is_absolute(), "destination FS path must be an absolut path."
Review Comment:
Nit: Typo `absolut`
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1220735232
##########
python/pyspark/sql/connect/session.py:
##########
@@ -637,6 +638,34 @@ def addArtifacts(self, *path: str, pyfile: bool = False, archive: bool = False)
addArtifact = addArtifacts
+ def copyFromLocalToFs(self, local_path: str, dest_path: str) -> None:
+ """
+ Copy file from local to cloud storage file system.
+ If the file already exits in destination path, old file is overwritten.
+
+ Parameters
+ ----------
+ local_path: str
+ Path to a local file (Directory is not supported yet).
+ The path can be either an absolute path or a relative path.
+
+ dest_path: str
+ The cloud storage path to the destination the file will
+ be copied to.
+ The path must be an an absolute path.
+
+ Notes
+ -----
+ This API is a developer API.
+ """
+ if urllib.parse.urlparse(dest_path).scheme:
+ raise ValueError(
+ "`spark_session.copyFromLocalToFs` API only allows `dest_path` to be a path "
+ "without scheme, and spark driver uses the default scheme to "
+ "determine the destination file system."
Review Comment:
Why can't we support the fully qualified URI with the scheme? We just add a prefix and truncate it later in the server.
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1220733748
##########
python/pyspark/sql/tests/connect/client/test_artifact.py:
##########
@@ -271,6 +277,21 @@ def func(x):
self.spark.addArtifacts(f"{archive_path}.zip#my_files", archive=True)
self.assertEqual(self.spark.range(1).select(func("id")).first()[0], "hello world!")
+ def test_copy_from_local_to_FS(self):
Review Comment:
```suggestion
def test_copy_from_local_to_fs(self):
```
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1219470038
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala:
##########
@@ -157,10 +159,46 @@ class SparkConnectArtifactManager private[connect] {
}
}
}
+
+ private[connect] def uploadArtifactToFs(
+ sessionHolder: SessionHolder,
+ remoteRelativePath: Path,
+ serverLocalStagingPath: Path): Unit = {
+ val hadoopConf = sessionHolder.session.sparkContext.hadoopConfiguration
+ assert(
+ remoteRelativePath.startsWith(
+ SparkConnectArtifactManager.forwardToFSPrefix + File.separator))
+ val destFSPath = new FSPath(
+ Paths
+ .get("/")
+ .resolve(remoteRelativePath.subpath(1, remoteRelativePath.getNameCount))
+ .toString)
+ val localPath = serverLocalStagingPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ if (fs.isInstanceOf[LocalFileSystem]) {
+ val allowDestLocalConf = sessionHolder.session.sparkContext.conf
+ .get("spark.fs.copyFromLocalToFs.allowDestLocal", "false")
Review Comment:
Used in testing.
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1220726346
##########
python/pyspark/sql/connect/session.py:
##########
@@ -637,6 +638,34 @@ def addArtifacts(self, *path: str, pyfile: bool = False, archive: bool = False)
addArtifact = addArtifacts
+ def copyFromLocalToFs(self, local_path: str, dest_path: str) -> None:
+ """
+ Copy file from local to cloud storage file system.
+ If the file already exits in destination path, old file is overwritten.
+
+ Parameters
+ ----------
+ local_path: str
+ Path to a local file (Directory is not supported yet).
Review Comment:
```suggestion
Path to a local file. Directories are not supported.
```
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1239306624
##########
python/pyspark/sql/connect/session.py:
##########
@@ -642,6 +643,34 @@ def addArtifacts(
addArtifact = addArtifacts
+ def copyFromLocalToFs(self, local_path: str, dest_path: str) -> None:
+ """
+ Copy file from local to cloud storage file system.
+ If the file already exits in destination path, old file is overwritten.
+
+ Parameters
+ ----------
+ local_path: str
+ Path to a local file. Directories are not supported.
+ The path can be either an absolute path or a relative path.
+
+ dest_path: str
+ The cloud storage path to the destination the file will
+ be copied to.
+ The path must be an an absolute path.
+
+ Notes
+ -----
+ This API is a developer API.
Review Comment:
Seems like we missed versionadded directive, @WeichenXu123
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYSPARK][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1209105558
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
+ try {
+ val hadoopConf = holder.session.sparkContext.hadoopConfiguration
+ assert(artifact.path.startsWith(forwardToFSPrefix))
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(artifact.path.subpath(1, artifact.path.getNameCount)).toString
+ )
+ val localPath = artifact.stagedPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
+ )
+ } catch {
+ case e: java.io.IOException =>
+ // TODO: report failure in response message
Review Comment:
Discussion: Shall we add a field in `AddArtifactsResponse` prototype message for holding failure information here ?
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212522454
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala:
##########
@@ -157,10 +159,34 @@ class SparkConnectArtifactManager private[connect] {
}
}
}
+
+ private[connect] def uploadArtifactToFs(
+ sessionHolder: SessionHolder,
+ remoteRelativePath: Path,
Review Comment:
I believe it's not a relative path?
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212527198
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala:
##########
@@ -157,10 +159,34 @@ class SparkConnectArtifactManager private[connect] {
}
}
}
+
+ private[connect] def uploadArtifactToFs(
+ sessionHolder: SessionHolder,
+ remoteRelativePath: Path,
+ serverLocalStagingPath: Path): Unit = {
+ val hadoopConf = sessionHolder.session.sparkContext.hadoopConfiguration
+ assert(remoteRelativePath.startsWith(
+ SparkConnectArtifactManager.forwardToFSPrefix + File.separator)
Review Comment:
I am fine with this but one thing is that now here we pass an absolute path with the prefix. Except here, all other places pass a relative path around. Is that fine, @hvanhovell and @vicennial?
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1211157822
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
+ try {
+ val hadoopConf = holder.session.sparkContext.hadoopConfiguration
+ assert(artifact.path.startsWith(forwardToFSPrefix))
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(artifact.path.subpath(1, artifact.path.getNameCount)).toString
+ )
+ val localPath = artifact.stagedPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
+ )
+ } catch {
+ case e: java.io.IOException =>
+ // TODO: report failure in response message
Review Comment:
`SparkConnectAddArtifactsHandler` already defined `onError` handler, see
https://github.com/apache/spark/blob/e8f87413bdd905e5e208ac8484b3b11de1623fc2/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala#L83
We might need to do some refactor to make it use `handleError` , @allisonwang-db would you mind to do it in follow-up PR ?
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1209592404
##########
python/pyspark/sql/connect/client/core.py:
##########
@@ -1240,6 +1240,9 @@ def _handle_rpc_error(self, rpc_error: grpc.RpcError) -> NoReturn:
def add_artifacts(self, *path: str, pyfile: bool, archive: bool) -> None:
self._artifact_manager.add_artifacts(*path, pyfile=pyfile, archive=archive)
+ def copy_from_local_to_FS(self, local_path: str, dest_path: str) -> None:
Review Comment:
`copy_from_local_to_FS` -> `copy_from_local_to_fs`
--
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
[GitHub] [spark] allisonwang-db commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "allisonwang-db (via GitHub)" <gi...@apache.org>.
allisonwang-db commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1210607108
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
+ try {
+ val hadoopConf = holder.session.sparkContext.hadoopConfiguration
+ assert(artifact.path.startsWith(forwardToFSPrefix))
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(artifact.path.subpath(1, artifact.path.getNameCount)).toString
+ )
+ val localPath = artifact.stagedPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
+ )
+ } catch {
+ case e: java.io.IOException =>
+ // TODO: report failure in response message
Review Comment:
I think we should catch all gRPC errors using this handleError: https://github.com/apache/spark/blob/7d87fecda7077714601cdf7328a11d2719a99bb1/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala#L119-L123
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFS` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1211157822
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala:
##########
@@ -89,6 +92,28 @@ class SparkConnectAddArtifactsHandler(val responseObserver: StreamObserver[AddAr
artifactManager.addArtifact(holder, artifact.path, artifact.stagedPath, artifact.fragment)
}
+ protected def uploadStagedArtifactToFS(artifact: StagedArtifact): Unit = {
+ try {
+ val hadoopConf = holder.session.sparkContext.hadoopConfiguration
+ assert(artifact.path.startsWith(forwardToFSPrefix))
+ val destFSPath = new FSPath(
+ Paths.get("/").resolve(artifact.path.subpath(1, artifact.path.getNameCount)).toString
+ )
+ val localPath = artifact.stagedPath
+ val fs = destFSPath.getFileSystem(hadoopConf)
+ fs.copyFromLocalFile(
+ false,
+ true,
+ new FSPath(localPath.toString),
+ destFSPath
+ )
+ } catch {
+ case e: java.io.IOException =>
+ // TODO: report failure in response message
Review Comment:
`SparkConnectAddArtifactsHandler` already defined `onError` handler (note this class inherits `StreamObserver`) , see
https://github.com/apache/spark/blob/e8f87413bdd905e5e208ac8484b3b11de1623fc2/connector/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectAddArtifactsHandler.scala#L83
We might need to do some refactor to make it use `handleError` , @allisonwang-db would you mind to do it in follow-up PR ?
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212521231
##########
connector/connect/server/src/test/scala/org/apache/spark/sql/connect/artifact/ArtifactManagerSuite.scala:
##########
@@ -145,4 +145,23 @@ class ArtifactManagerSuite extends SharedSparkSession with ResourceHelper {
assert(artifactManager.getSparkConnectPythonIncludes == Seq("abc.zip"))
}
}
+
+ test("Forward artifact file to cloud storage path") {
Review Comment:
```suggestion
test("SPARK-43790: Forward artifact file to cloud storage path") {
```
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "HyukjinKwon (via GitHub)" <gi...@apache.org>.
HyukjinKwon commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212527926
##########
python/pyspark/sql/connect/client/artifact.py:
##########
@@ -187,6 +188,19 @@ def _parse_artifacts(self, path_or_uri: str, pyfile: bool, archive: bool) -> Lis
return [artifact]
raise RuntimeError(f"Unsupported scheme: {parsed.scheme}")
+ def _parse_forward_to_FS_artifacts(self, local_path: str, dest_path: str) -> List[Artifact]:
Review Comment:
BTW, this one too. Shouldn't we do it together with `_parse_artifacts`?
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212561246
##########
connector/connect/server/src/main/scala/org/apache/spark/sql/connect/artifact/SparkConnectArtifactManager.scala:
##########
@@ -157,10 +159,34 @@ class SparkConnectArtifactManager private[connect] {
}
}
}
+
+ private[connect] def uploadArtifactToFs(
+ sessionHolder: SessionHolder,
+ remoteRelativePath: Path,
+ serverLocalStagingPath: Path): Unit = {
+ val hadoopConf = sessionHolder.session.sparkContext.hadoopConfiguration
+ assert(remoteRelativePath.startsWith(
+ SparkConnectArtifactManager.forwardToFSPrefix + File.separator)
Review Comment:
In the design suggested by @vicennial , We defined the artifact path to be `forward_to_fs/{dest_fs_path}`, and `dest_fs_path` is absolute path to cloud storage. I think it is fine and it works in all cases.
--
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
[GitHub] [spark] WeichenXu123 commented on a diff in pull request #41357: [SPARK-43790][PYTHON][CONNECT][ML] Add `copyFromLocalToFs` API
Posted by "WeichenXu123 (via GitHub)" <gi...@apache.org>.
WeichenXu123 commented on code in PR #41357:
URL: https://github.com/apache/spark/pull/41357#discussion_r1212765607
##########
python/pyspark/sql/connect/session.py:
##########
@@ -625,6 +625,26 @@ def addArtifacts(self, *path: str, pyfile: bool = False, archive: bool = False)
addArtifact = addArtifacts
+ def copyFromLocalToFs(self, local_path: str, dest_path: str) -> None:
Review Comment:
I feel like this API is useful in many places, not only for ML, what about keeps putting it here ? We can still mark it as developer API
--
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