数据流 如何将 PubSub 消息写入 GCS 单独文件

Question

我有一个场景，其中有一个流数据流。逻辑是从 PubSub 订阅中读取 json 数据 并对外部系统进行 api 调用。如果外部系统无法访问，我需要将 json 数据保存到 GCS 以便以后重新处理。我可以将故障数据写入 GCS，但我面临的挑战是，它不是在提到的窗口之间写入单个文件，而是写入单个文件。有什么方法可以根据每个 json 数据的值将每个 pubsub 消息作为具有唯一文件名的单独文件写入 GCS。

    final TupleTag<String> failureData =new TupleTag<String>(){};
    PCollectionTuple APIOutputCollection= pipeline.apply("Read PubSub Events",PubsubIO.readMessagesWithAttributes().fromSubscription(options.getInputSubscription()))
           .apply("Send to Apigee",ParDo.of(new APICall()).withOutputTags(successData,TupleTagList.of(failureData)));
    
    
     APIOutputCollection.get(failureData)
     .apply(
                options.getWindowDuration() + " Window",
                Window.into(FixedWindows.of(DurationUtils.parseDuration(options.getWindowDuration()))))
        .apply(
                "Write File(s)",
                TextIO.write()
                    .withWindowedWrites()
                    .withNumShards(options.getNumShards())
                    .to(
                        WindowedFilenamePolicy.writeWindowedFiles()
                            .withOutputDirectory(options.getOutputDir())
                            .withOutputFilenamePrefix((options.getOutputFileNamePrefix())
                            .withShardTemplate(options.getOutputShardTemplate())
                            .withSuffix(".json")
                            .withTempDirectory(
                            NestedValueProvider.of(
                            maybeUseUserTempLocation(options.getUserTempLocation(),options.getOutputDirectory()),
                           (SerializableFunction<String, ResourceId>)
                           input -> FileBasedSink.convertToFileResourceIfPossible(input))));