将数据流从 Google Cloud Storage 流式传输到 Big Query

Question

我正在尝试使用 DataFlow (Java) 将数据从 Cloud Storage 插入到 Big Query。我可以批量上传数据；但是，我想改为设置流式上传。因此，当新对象添加到我的存储桶时，它们将被推送到 BigQuery。

我已将 PipelineOptions 设置为 Streaming，它在 GCP Console UI 中显示数据流管道属于流式传输类型。我在存储桶中的初始文件/对象集被推送到 BigQuery。

但是当我向存储桶添加新对象时，这些对象不会被推送到 BigQuery。这是为什么？如何使用流式数据流管道将添加到我的 Cloud Storage 的对象推送到 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);

我的解释是，因为这是一个流式管道，所以当我将对象添加到 Cloud Storage 时，它​​们会被推送到 BigQuery。

请提出建议。

Answer 1

您如何创建输入集合？您需要有一个无限制的输入才能使流式传输管道继续运行，否则它将只是暂时的（但将使用流式插入）。 您可以通过读取包含存储桶中所有更改的订阅来实现此目的，请参阅https://cloud.google.com/storage/docs/pubsub-notifications 了解详细信息。