Basically I am reading data from pubsub and writing data to google storage. Code snippet is as below.
public class WriteWindowedFile extends PTransform<PCollection<String>, PDone> {
private String bucketLocation;
private LogTypeEnum logTypeEnum;
private int shards;
public WriteWindowedFile(String bucketLocation, LogTypeEnum logTypeEnum, int shards) {
this.bucketLocation = bucketLocation;
this.logTypeEnum = logTypeEnum;
this.shards = shards;
}
@Override
public PDone expand(PCollection<String> input) {
checkArgument(input.getWindowingStrategy().getWindowFn().windowCoder() == IntervalWindow.getCoder());
ResourceId resource = FileBasedSink.convertToFileResourceIfPossible(bucketLocation);
return input.apply(
TextIO.write()
.to(new FileStorageFileNamePolicy(logTypeEnum))
.withTempDirectory(resource.getCurrentDirectory())
.withWindowedWrites()
.withNumShards(shards)
);
}
}
The FilenamePolicy implementation is:
public class FileStorageFileNamePolicy extends FileBasedSink.FilenamePolicy {
private static final long serialVersionUID = 1L;
private static final Logger LOGGER = LoggerFactory.getLogger(FileStorageFileNamePolicy.class);
private LogTypeEnum logTypeEnum;
public FileStorageFileNamePolicy(LogTypeEnum logTypeEnum) {
this.logTypeEnum = logTypeEnum;
}
@Override
public ResourceId windowedFilename(int shardNumber,
int numShards,
BoundedWindow window,
PaneInfo paneInfo,
FileBasedSink.OutputFileHints outputFileHints) {
IntervalWindow intervalWindow = (IntervalWindow) window;
String startDate = intervalWindow.start().toString();
String dateString = startDate.replace("T", CommonConstants.SPACE)
.replaceAll(startDate.substring(startDate.indexOf('Z')), CommonConstants.EMPTY_STRING);
try {
startDate = DateUtil.getDateForFileStore(dateString, null);
} catch (ParseException e) {
LOGGER.error("Error converting date : {}", e);
}
String filename = intervalWindow.start().toString() + ".txt";
String dirName = startDate + CommonConstants.FORWARD_SLASH +
logTypeEnum.getValue().toLowerCase() + CommonConstants.FORWARD_SLASH;
LOGGER.info("Directory : {} and File Name : {}", dirName, filename);
return FileBasedSink.convertToFileResourceIfPossible(filename).
resolve(dirName, ResolveOptions.StandardResolveOptions.RESOLVE_DIRECTORY);
}
@Nullable
@Override
public ResourceId unwindowedFilename(
int shardNumber, int numShards, FileBasedSink.OutputFileHints outputFileHints) {
throw new UnsupportedOperationException("Unsupported");
}
}
While writing to google storage I am facing following issue even when I am passing the actual directory path. Getting below stacktrace when it tries to resolve the directory in FileStorageFileNamePolicy class.
exception: "java.lang.RuntimeException: org.apache.beam.sdk.util.UserCodeException: java.lang.IllegalStateException: Expected the path is a directory, but had [/2019-09-23T16:59:42.189Z.txt]. at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn$1.output(GroupAlsoByWindowsParDoFn.java:184) at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner$1.outputWindowedValue(GroupAlsoByWindowFnRunner.java:102) at org.apache.beam.runners.dataflow.worker.StreamingGroupAlsoByWindowReshuffleFn.processElement(StreamingGroupAlsoByWindowReshuffleFn.java:57) at org.apache.beam.runners.dataflow.worker.StreamingGroupAlsoByWindowReshuffleFn.processElement(StreamingGroupAlsoByWindowReshuffleFn.java:39) at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.invokeProcessElement(GroupAlsoByWindowFnRunner.java:115) at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.processElement(GroupAlsoByWindowFnRunner.java:73) at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn.processElement(GroupAlsoByWindowsParDoFn.java:134) at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44) at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:49) at org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:201) at org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.start(ReadOperation.java:159) at org.apache.beam.runners.dataflow.worker.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:77) at org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1295) at org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.access$1000(StreamingDataflowWorker.java:149) at org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker$6.run(StreamingDataflowWorker.java:1028) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) at java.lang.Thread.run(Thread.java:745) Caused by: org.apache.beam.sdk.util.UserCodeException: java.lang.IllegalStateException: Expected the path is a directory, but had [/2019-09-23T16:59:42.189Z.txt]. at org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:34) at org.apache.beam.sdk.io.WriteFiles$FinalizeTempFileBundles$FinalizeFn$DoFnInvoker.invokeProcessElement(Unknown Source) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:214) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:179) at org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement(SimpleParDoFn.java:330) at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44) at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:49) at org.apache.beam.runners.dataflow.worker.SimpleParDoFn$1.output(SimpleParDoFn.java:276) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue(SimpleDoFnRunner.java:248) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$700(SimpleDoFnRunner.java:74) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:560) at org.apache.beam.sdk.transforms.DoFnOutputReceivers$WindowedContextOutputReceiver.output(DoFnOutputReceivers.java:71) at org.apache.beam.sdk.transforms.MapElements$1.processElement(MapElements.java:139) at org.apache.beam.sdk.transforms.MapElements$1$DoFnInvoker.invokeProcessElement(Unknown Source) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:214) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:179) at org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement(SimpleParDoFn.java:330) at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44) at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:49) at org.apache.beam.runners.dataflow.worker.SimpleParDoFn$1.output(SimpleParDoFn.java:276) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue(SimpleDoFnRunner.java:248) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$700(SimpleDoFnRunner.java:74) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:560) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:548) at org.apache.beam.runners.dataflow.ReshuffleOverrideFactory$ReshuffleWithOnlyTrigger$1.processElement(ReshuffleOverrideFactory.java:86) at org.apache.beam.runners.dataflow.ReshuffleOverrideFactory$ReshuffleWithOnlyTrigger$1$DoFnInvoker.invokeProcessElement(Unknown Source) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:214) at org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:179) at org.apache.beam.runners.dataflow.worker.SimpleParDoFn.processElement(SimpleParDoFn.java:330) at org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44) at org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:49) at org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn$1.output(GroupAlsoByWindowsParDoFn.java:182) ... 17 more Caused by: java.lang.IllegalStateException: Expected the path is a directory, but had [/2019-09-23T16:59:42.189Z.txt]. at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkState(Preconditions.java:588) at org.apache.beam.sdk.io.LocalResourceId.resolve(LocalResourceId.java:57) at org.apache.beam.sdk.io.LocalResourceId.resolve(LocalResourceId.java:36) at com.vuclip.dataflow.pipeline.helper.FileStorageFileNamePolicy.windowedFilename(FileStorageFileNamePolicy.java:54) at org.apache.beam.sdk.io.FileBasedSink$FileResult.getDestinationFile(FileBasedSink.java:1086) at org.apache.beam.sdk.io.FileBasedSink$WriteOperation.finalizeDestination(FileBasedSink.java:645) at org.apache.beam.sdk.io.WriteFiles.finalizeAllDestinations(WriteFiles.java:872) at org.apache.beam.sdk.io.WriteFiles.access$1600(WriteFiles.java:111) at org.apache.beam.sdk.io.WriteFiles$FinalizeTempFileBundles$FinalizeFn.process(WriteFiles.java:849)
Can anyone please help ? Thanks
