java - 我们可以将avro文件写入动态创建的GCS存储桶(基于tenantID)吗？

Question

基本上我想做的是，根据tenantID(作为事件的一部分)创建GCS存储桶，并使用Google数据流作业中的动态文件命名使用FileIO.writeDynamic写入这些事件。

我面临的问题是

srcEvents.apply("Window", Window
                        .<MyEvent>into(FixedWindows.of(Duration.standardSeconds(60))))
                        .apply("WriteAvro", FileIO.<MyEventDestination, MyEvent>writeDynamic()
                                        .by(groupFn).via(outputFn, sinkFn)
                                        **.to()** // what to pass as here as i want it to be based on event.getTenantId (gs://test-123)
                                        .withDestinationCoder(destinationCoder)
                                        .withNumShards(100).withNaming(namingFn));

我在上述之前通过调用 PTranform of srcEvents 创建 gcs 存储桶

Answer 1

我能够通过使用 withTempDirectory 选项解决此问题，其中我提供了临时 gcs 存储桶路径并使用文件命名来构建每个域的动态存储桶路径

srcEvents.apply("Window", Window
.<MyEvent>into(FixedWindows.of(Duration.standardSeconds(60))))
.apply("WriteAvro", FileIO.<MyEventDestination, MyEvent>writeDynamic()
.by(groupFn).via(outputFn, sinkFn)
.withTempDirectory("gs://temp-blah/")
.withDestinationCoder(destinationCoder)
.withNumShards(100).withNaming(namingFn));
namingFn to build filename gs://domain-123/2020-05-01/event.avro