You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "liferoad (via GitHub)" <gi...@apache.org> on 2024/11/02 03:07:01 UTC
[PR] Added replaceGcsFilesWithLocalFiles [beam]
liferoad opened a new pull request, #33006:
URL: https://github.com/apache/beam/pull/33006
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added replaceGcsFilesWithLocalFiles [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #33006:
URL: https://github.com/apache/beam/pull/33006#discussion_r1842168716
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java:
##########
@@ -257,6 +260,92 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> {
/** Dataflow service endpoints are expected to match this pattern. */
static final String ENDPOINT_REGEXP = "https://[\\S]*googleapis\\.com[/]?";
+ /**
+ * Replaces GCS file paths with local file paths by downloading the GCS files locally. This is
+ * useful when files need to be accessed locally before being staged to Dataflow.
+ *
+ * @param filesToStage List of file paths that may contain GCS paths (gs://) and local paths
+ * @return List of local file paths where any GCS paths have been downloaded locally
+ * @throws RuntimeException if there are errors copying GCS files locally
+ */
+ public static List<String> replaceGcsFilesWithLocalFiles(List<String> filesToStage) {
+ List<String> processedFiles = new ArrayList<>();
+
+ for (String fileToStage : filesToStage) {
+ String localPath;
+ if (fileToStage.contains("=")) {
+ // Handle files with staging name specified
+ String[] components = fileToStage.split("=", 2);
+ String stagingName = components[0];
+ String filePath = components[1];
+
+ if (filePath.startsWith("gs://")) {
+ try {
+ // Create temp file with exact same name as GCS file
+ String gcsFileName = filePath.substring(filePath.lastIndexOf('/') + 1);
+ File tempDir = Files.createTempDir();
+ tempDir.deleteOnExit();
+ File tempFile = new File(tempDir, gcsFileName);
+ tempFile.deleteOnExit();
+
+ LOG.info(
+ "Downloading GCS file {} to local temp file {}",
+ filePath,
+ tempFile.getAbsolutePath());
+
+ // Copy GCS file to local temp file
+ ResourceId source = FileSystems.matchNewResource(filePath, false);
+ try (ReadableByteChannel reader = FileSystems.open(source);
+ FileOutputStream writer = new FileOutputStream(tempFile)) {
+ ByteStreams.copy(Channels.newInputStream(reader), writer);
Review Comment:
What you implemented is just the same as `copy` but now we have two copies to maintain, really. The layers should be robust.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added replaceGcsFilesWithLocalFiles [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on PR #33006:
URL: https://github.com/apache/beam/pull/33006#issuecomment-2479362311
Will close this PR and create a new one to do this when staging the GCS files.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added replaceGcsFilesWithLocalFiles [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #33006:
URL: https://github.com/apache/beam/pull/33006#discussion_r1847462508
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java:
##########
@@ -257,6 +260,92 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> {
/** Dataflow service endpoints are expected to match this pattern. */
static final String ENDPOINT_REGEXP = "https://[\\S]*googleapis\\.com[/]?";
+ /**
+ * Replaces GCS file paths with local file paths by downloading the GCS files locally. This is
+ * useful when files need to be accessed locally before being staged to Dataflow.
+ *
+ * @param filesToStage List of file paths that may contain GCS paths (gs://) and local paths
+ * @return List of local file paths where any GCS paths have been downloaded locally
+ * @throws RuntimeException if there are errors copying GCS files locally
+ */
+ public static List<String> replaceGcsFilesWithLocalFiles(List<String> filesToStage) {
+ List<String> processedFiles = new ArrayList<>();
+
+ for (String fileToStage : filesToStage) {
+ String localPath;
+ if (fileToStage.contains("=")) {
+ // Handle files with staging name specified
+ String[] components = fileToStage.split("=", 2);
+ String stagingName = components[0];
+ String filePath = components[1];
+
+ if (filePath.startsWith("gs://")) {
+ try {
+ // Create temp file with exact same name as GCS file
+ String gcsFileName = filePath.substring(filePath.lastIndexOf('/') + 1);
+ File tempDir = Files.createTempDir();
+ tempDir.deleteOnExit();
+ File tempFile = new File(tempDir, gcsFileName);
+ tempFile.deleteOnExit();
+
+ LOG.info(
+ "Downloading GCS file {} to local temp file {}",
+ filePath,
+ tempFile.getAbsolutePath());
+
+ // Copy GCS file to local temp file
+ ResourceId source = FileSystems.matchNewResource(filePath, false);
+ try (ReadableByteChannel reader = FileSystems.open(source);
+ FileOutputStream writer = new FileOutputStream(tempFile)) {
+ ByteStreams.copy(Channels.newInputStream(reader), writer);
Review Comment:
```
Expect srcResourceIds and destResourceIds have the same scheme, but received gs, file.
java.lang.IllegalArgumentException: Expect srcResourceIds and destResourceIds have the same scheme, but received gs, file.
```
Looks like our copy needs the same scheme. :)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added replaceGcsFilesWithLocalFiles [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on PR #33006:
URL: https://github.com/apache/beam/pull/33006#issuecomment-2484433958
some offline discussions with @kennknowles. Adding this to the staging phase needs more time and more places to be changed since the current staging logic relies on local files (e.g., computing hash). So we decided to merge this PR for now. The improvements on the staging phase could be done later.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added replaceGcsFilesWithLocalFiles [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #33006:
URL: https://github.com/apache/beam/pull/33006#issuecomment-2457588065
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @kennknowles added as fallback since no labels match configuration
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added replaceGcsFilesWithLocalFiles [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad commented on code in PR #33006:
URL: https://github.com/apache/beam/pull/33006#discussion_r1841128358
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java:
##########
@@ -257,6 +260,92 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> {
/** Dataflow service endpoints are expected to match this pattern. */
static final String ENDPOINT_REGEXP = "https://[\\S]*googleapis\\.com[/]?";
+ /**
+ * Replaces GCS file paths with local file paths by downloading the GCS files locally. This is
+ * useful when files need to be accessed locally before being staged to Dataflow.
+ *
+ * @param filesToStage List of file paths that may contain GCS paths (gs://) and local paths
+ * @return List of local file paths where any GCS paths have been downloaded locally
+ * @throws RuntimeException if there are errors copying GCS files locally
+ */
+ public static List<String> replaceGcsFilesWithLocalFiles(List<String> filesToStage) {
+ List<String> processedFiles = new ArrayList<>();
+
+ for (String fileToStage : filesToStage) {
+ String localPath;
+ if (fileToStage.contains("=")) {
+ // Handle files with staging name specified
+ String[] components = fileToStage.split("=", 2);
+ String stagingName = components[0];
+ String filePath = components[1];
+
+ if (filePath.startsWith("gs://")) {
+ try {
+ // Create temp file with exact same name as GCS file
+ String gcsFileName = filePath.substring(filePath.lastIndexOf('/') + 1);
+ File tempDir = Files.createTempDir();
+ tempDir.deleteOnExit();
+ File tempFile = new File(tempDir, gcsFileName);
+ tempFile.deleteOnExit();
+
+ LOG.info(
+ "Downloading GCS file {} to local temp file {}",
+ filePath,
+ tempFile.getAbsolutePath());
+
+ // Copy GCS file to local temp file
+ ResourceId source = FileSystems.matchNewResource(filePath, false);
+ try (ReadableByteChannel reader = FileSystems.open(source);
+ FileOutputStream writer = new FileOutputStream(tempFile)) {
+ ByteStreams.copy(Channels.newInputStream(reader), writer);
Review Comment:
Our `Filesystems.copy` has too many layers. Here, we simply downloads the file to a given temp file, which is also removed with `deleteOnExit`. I think we do not need to use our internal tool since it is quite straightforward.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added replaceGcsFilesWithLocalFiles [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #33006:
URL: https://github.com/apache/beam/pull/33006#discussion_r1834891646
##########
runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java:
##########
@@ -257,6 +260,92 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> {
/** Dataflow service endpoints are expected to match this pattern. */
static final String ENDPOINT_REGEXP = "https://[\\S]*googleapis\\.com[/]?";
+ /**
+ * Replaces GCS file paths with local file paths by downloading the GCS files locally. This is
+ * useful when files need to be accessed locally before being staged to Dataflow.
+ *
+ * @param filesToStage List of file paths that may contain GCS paths (gs://) and local paths
+ * @return List of local file paths where any GCS paths have been downloaded locally
+ * @throws RuntimeException if there are errors copying GCS files locally
+ */
+ public static List<String> replaceGcsFilesWithLocalFiles(List<String> filesToStage) {
+ List<String> processedFiles = new ArrayList<>();
+
+ for (String fileToStage : filesToStage) {
+ String localPath;
+ if (fileToStage.contains("=")) {
+ // Handle files with staging name specified
+ String[] components = fileToStage.split("=", 2);
+ String stagingName = components[0];
+ String filePath = components[1];
+
+ if (filePath.startsWith("gs://")) {
+ try {
+ // Create temp file with exact same name as GCS file
+ String gcsFileName = filePath.substring(filePath.lastIndexOf('/') + 1);
+ File tempDir = Files.createTempDir();
+ tempDir.deleteOnExit();
+ File tempFile = new File(tempDir, gcsFileName);
+ tempFile.deleteOnExit();
+
+ LOG.info(
+ "Downloading GCS file {} to local temp file {}",
+ filePath,
+ tempFile.getAbsolutePath());
+
+ // Copy GCS file to local temp file
+ ResourceId source = FileSystems.matchNewResource(filePath, false);
+ try (ReadableByteChannel reader = FileSystems.open(source);
+ FileOutputStream writer = new FileOutputStream(tempFile)) {
+ ByteStreams.copy(Channels.newInputStream(reader), writer);
Review Comment:
I don't totally understand the use of this method, but surely you should use `Filesystems.copy` method that does this: https://github.com/apache/beam/blob/159c1a4f67aa8b634fdb4ab03afde432ce8ed0d2/sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileSystems.java#L276
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added replaceGcsFilesWithLocalFiles [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad closed pull request #33006: Added replaceGcsFilesWithLocalFiles
URL: https://github.com/apache/beam/pull/33006
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added replaceGcsFilesWithLocalFiles [beam]
Posted by "liferoad (via GitHub)" <gi...@apache.org>.
liferoad merged PR #33006:
URL: https://github.com/apache/beam/pull/33006
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org