基于元素值的Google云存储数据流写入

Question

我试图构建一个数据流过程，通过将数据存储到谷歌云存储中来帮助归档数据。我有一个事件数据的PubSub流，其中包含client_id和一些元数据。这个进程应该归档所有传入的事件，因此这需要是一个流管道。

我希望能够通过将接收到的每个事件放入类似gs://archive/client_id/eventdata.json的bucket中来处理事件归档。在DataFlow/Apache beam中是否可以这样做，特别是能够为pCollection中的每个事件分配不同的文件名？

编辑：所以我的代码当前看起来像：

public static class PerWindowFiles extends FileBasedSink.FilenamePolicy {

private String customerId;

public PerWindowFiles(String customerId) {
  this.customerId = customerId;
}

@Override
public ResourceId windowedFilename(ResourceId outputDirectory, WindowedContext context, String extension) {
  String filename = bucket+"/"+customerId;
  return outputDirectory.resolve(filename, ResolveOptions.StandardResolveOptions.RESOLVE_FILE);
}

@Override
public ResourceId unwindowedFilename(
    ResourceId outputDirectory, Context context, String extension) {
  throw new UnsupportedOperationException("Unsupported.");
}
}


public static void main(String[] args) throws IOException {
DataflowPipelineOptions options = PipelineOptionsFactory.fromArgs(args)
    .withValidation()
    .as(DataflowPipelineOptions.class);
options.setRunner(DataflowRunner.class);
options.setStreaming(true);
Pipeline p = Pipeline.create(options);

PCollection<Event> set = p.apply(PubsubIO.readStrings()
                                     .fromTopic("topic"))
    .apply(new ConvertToEvent()));

PCollection<KV<String, Event>> events = labelEvents(set);
PCollection<KV<String, EventGroup>> sessions = groupEvents(events);

String customers = System.getProperty("CUSTOMERS");
JSONArray custList = new JSONArray(customers);
for (Object cust : custList) {
  if (cust instanceof String) {
    String customerId = (String) cust;
    PCollection<KV<String, EventGroup>> custCol = sessions.apply(new FilterByCustomer(customerId));
            stringifyEvents(custCol)
                .apply(TextIO.write()
                                               .to("gs://archive/")
                                               .withFilenamePolicy(new PerWindowFiles(customerId))
                                               .withWindowedWrites()
                                               .withNumShards(3));
  } else {
    LOG.info("Failed to create TextIO: customerId was not String");
  }
}

p.run()
    .waitUntilFinish();
}

这段代码很难看，因为每次发生新的客户机时，我都需要重新部署，以便能够保存它们的数据。我更希望能够动态地将客户数据分配到适当的桶中。

Answer 1

“Dynamic Destinations”--根据正在编写的元素选择文件名--将是Beam 2.1.0中的一个新功能，该功能尚未发布。