我已经看到了很多关于同一主题的问题。但是，我仍然有问题写信给GCS。我正在阅读pubsub的主题，并试图将其推送到GCS。我参考了这个链接。但是，在最新的光束包中找不到IOChannelUtils。
PCollection<String> details = pipeline
.apply(PubsubIO.readStrings().fromTopic("/topics/<project>/sampleTopic"));
PCollection<KV<String, String>> keyedStream = details.apply(WithKeys.of(new SerializableFunction<String, String>() {
public String apply(String s) {
return "constant";
}
}));
PCollection<KV<String, Iterable<String>>> keyedWindows = keyedStream.apply(Window.<KV<String, String>>into(FixedWindows.of(ONE_MIN)).withAllowedLateness(ONE_DAY)
.triggering(AfterWatermark.pastEndOfWindow().withEarlyFirings(AfterPane.elementCountAtLeast(10))
.withLateFirings(AfterFirst.of(AfterPane.elementCountAtLeast(10),
AfterProcessingTime.pastFirstElementInPane().plusDelayOf(TEN_SECONDS))))
.discardingFiredPanes()).apply(GroupByKey.create());
PCollection<Iterable<String>> windows = keyedWindows.apply(Values.create());
这是我从堆栈溢出中的许多其他类似主题中获得的。现在，我明白了，TextIO确实支持与WindowedWrites和与NumShards一起使用的无界PCollection写入选项。
参考：使用DoFn使用Cloud Dataflow从PubSub写入Google Cloud Storage
但是，我不明白，我应该怎么做。
我试图写信给GCS如下。
FilenamePolicy policy = DefaultFilenamePolicy.constructUsingStandardParameters(
StaticValueProvider.of(outputDirectory), DefaultFilenamePolicy.DEFAULT_SHARD_TEMPLATE, "");
details.apply(TextIO.write().to("gs://<bucket>/topicfile").withWindowedWrites()
.withFilenamePolicy(policy).withNumShards(4));
我没有足够的观点来为Stack Overflow中的这些主题添加评论，因此我将其作为一个不同的问题提出。
我可以通过修改窗口来解决这个问题，如下所示
PCollection<String> streamedDataWindows = streamedData.apply(Window.<String>into(new GlobalWindows())
.triggering(Repeatedly
.forever(AfterProcessingTime
.pastFirstElementInPane()
.plusDelayOf(Duration.standardSeconds(30))
)).withAllowedLateness(Duration.standardDays(1)).discardingFiredPanes());
streamedDataWindows.apply(TextIO.write().to(CLOUD_STORAGE).withWindowedWrites().withNumShards(1).withFilenamePolicy(new PerWindowFiles()));
public static class PerWindowFiles extends FileBasedSink.FilenamePolicy {
public ResourceId windowedFilename(ResourceId outputDirectory, WindowedContext context, String extension) {
// OVERRIDE THE FILE NAME CREATION
}
}
虽然我可以这样解决它，但我仍然不确定这里的窗口概念。当我找到它时，我会添加更多细节。如果有人有更多的理解，请添加更多细节。谢谢
查看此Pub/Sub toGCSPipeline，它提供了将窗口文件写入GCS的完整示例。
