FileBasedSink writing to incorrect path when path prefixe has no file component in path (e.g. /tmp/)
- Dominant language
- Java
- Stars
- 8.7k
- Forks
- 4.7k
- Avg merge
- 2d 2h
- Merged PRs (30d)
- 205
Description
This was reported by a user on dev@, original report:
https://lists.apache.org/thread.html/378da40ca7d13e226ca793d2a27af047f9a562f273f6eaa5d677dc4b@%3Cdev.beam.apache.org%3E
This pipeline (where WindowedFilenamePolicy is the one found in org.apache.beam.sdk.io.AvroIOTest) produces files in the wrong directory:
```
stringsPCollection.apply(Window.into(FixedWindows.of(Duration.standardSeconds(5))))
.apply("WritingToOutput",
TextIO.write().withWindowedWrites().withFilenamePolicy(new
WindowedFilenamePolicy("my_pref")).to("/tmp/").withNumShards(1));
```
I expected the output files to be written to /tmp/ directory but they are not. They are written to root directory which is unexpected behavior.
I think the problem is org.apache.beam.sdk.io.FileBasedSink.ExtractDirectory ...
This main method shows the problem:
```
import org.apache.beam.sdk.io.FileBasedSink;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.options.ValueProvider;
import
org.apache.beam.sdk.options.ValueProvider.StaticValueProvider;
public class Test {
public static
void main(String[] args) {
ResourceId res = FileBasedSink.convertToFileResourceIfPossible("/tmp/");
System.out.println("Resource is " + res + ", current directory is " +
res.getCurrentDirectory() +
", filename is " + res.getFilename());
FileBasedSink mockFBS = new
FileBasedSink(StaticValueProvider.of(res),
null) {
@Override
public org.apache.beam.sdk.io.FileBasedSink.WriteOperation
createWriteOperation()
{
return null;
}
};
final ValueProvider provider =
mockFBS.getBaseOutputDirectoryProvider();
System.out.println("BaseOutputProvider is " + provider + ",
isAccessible=" + provider.isAccessible()
+ ", getValue=" + provider.get());
}
}
```
The output is
```
Resource is /tmp, current directory is //, filename is tmp
BaseOutputProvider is
NestedValueProvider{value=StaticValueProvider{value=/tmp}},
isAccessible=true,
getValue=//
```
Imported from Jira [BEAM-2448](https://issues.apache.org/jira/browse/BEAM-2448). Original Jira may contain additional context.
Reported by: lcwik.
Contributor guide
Research direction
Start with org.apache.beam.sdk.io.FileBasedSink, especially ExtractDirectory, and reproduce the behavior using the Java example in the issue with the /tmp/ path prefix. Done means the base output directory resolves to /tmp/ rather than the root directory when the prefix has no file component.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100