从Google云存储流数据流到Big Query

Question

我正在尝试使用DataFlow（Java）将数据从云存储插入到Big Query中。我可以批量上传数据；但是，我想要设置一个流式上传代替。因此，当新对象添加到我的bucket时，它们将被推送到BigQuery。

我已经将PipelineOptions设置为流，并且在GCP控制台UI中显示dataflow管道是流类型的。bucket中的初始文件/对象集被推送到BigQuery。

但是当我向桶中添加新对象时，这些对象不会被推送到BigQuery。这是为什么？我如何使用蒸汽数据流管道将添加到我的云存储中的对象推送到BigQuery？

//Specify PipelineOptions
DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);


  options.setProject(<project-name>);
  options.setStagingLocation(<bucket/staging folder>);    
  options.setStreaming(true);
  options.setRunner(DataflowRunner.class);

我的解释是，因为这是一个流式管道，当我向云存储添加对象时，它们将被推送到BigQuery。

请建议。

Answer 1

如何创建输入集合？您需要有一个无界输入，以便流管道保持打开状态，否则它将只是临时的（但将使用流插入）。您可以通过从包含桶中所有更改的订阅中读取内容来实现这一点，请参阅https://cloud.google.com/storage/docs/pubsub-notifications了解详细信息。