从 Google Cloud BigQuery 读取数据

Jan*_*Jan 1 java google-app-engine dataflow google-bigquery google-cloud-dataflow

我是 Pipeline 世界和 Google API DataFlow 的新手。

我想使用 sqlQuery 从 BigQuery 读取数据。当我读取所有数据库时,它工作正常。

PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
PCollection<TableRow> qData = p.apply(
     BigQueryIO.Read
         .named("Read")
         .from("test:DataSetTest.data"));
Run Code Online (Sandbox Code Playgroud)

但是当我使用 fromQuery 时出现错误。

PipelineOptions options = PipelineOptionsFactory.create();
Pipeline p = Pipeline.create(options);
PCollection<TableRow> qData = p.apply(
     BigQueryIO.Read
         .named("Read")
         .fromQuery("SELECT * FROM DataSetTest.data"));
Run Code Online (Sandbox Code Playgroud)

错误:

线程“main”中出现异常 java.lang.IllegalArgumentException:查询“SELECT * FROM DataSetTest.data”的验证失败。如果查询依赖于管道的早期阶段,则可以使用#withoutValidation 禁用此验证。

在 com.google.cloud.dataflow.sdk.io.BigQueryIO$Read$Bound.dryRunQuery(BigQueryIO.java:449)

在 com.google.cloud.dataflow.sdk.io.BigQueryIO$Read$Bound.validate(BigQueryIO.java:432)

在com.google.cloud.dataflow.sdk.Pipeline.applyInternal(Pipeline.java:357)

在com.google.cloud.dataflow.sdk.Pipeline.applyTransform(Pipeline.java:267)

在 com.google.cloud.dataflow.sdk.values.PBegin.apply(PBegin.java:47)

在com.google.cloud.dataflow.sdk.Pipeline.apply(Pipeline.java:151)

在 Test.java.packageid.StarterPipeline.main(StarterPipeline.java:72)

引起原因:java.lang.NullPointerException:必须指定必需参数projectId。

在 com.google.api.client.repackaged.com.google.common.base.Preconditions.checkNotNull(Preconditions.java:229)

在 com.google.api.client.util.Preconditions.checkNotNull(Preconditions.java:140)

在 com.google.api.services.bigquery.Bigquery$Jobs$Query。(Bigquery.java:1751)

在 com.google.api.services.bigquery.Bigquery$Jobs.query(Bigquery.java:1724)

在 com.google.cloud.dataflow.sdk.io.BigQueryIO$Read$Bound.dryRunQuery(BigQueryIO.java:445)

... 6 更多

这里有什么问题?

更新:

我通过“options.setProject”设置项目。

PipelineOptions options = PipelineOptionsFactory.create();
    Pipeline p = Pipeline.create(options);
    options.setProject("test");
    PCollection<TableRow> qData = p.apply(
         BigQueryIO.Read
             .named("Read")
             .fromQuery("SELECT * FROM DataSetTest.data"));
Run Code Online (Sandbox Code Playgroud)

但现在我收到了这个消息。未找到表。

引起原因:com.google.api.client.googleapis.json.GoogleJsonResponseException:404未找到{“代码”:404,“错误”:[{“域”:“全局”,“消息”:“未找到:表测试:_dataflow_temporary_dataset_737099.dataflow_temporary_table_550832”,“原因”:“notFound”}],“消息”:“未找到:表测试:_dataflow_temporary_dataset_737099.dataflow_temporary_table_550832”}

Dav*_*aci 6

Google Cloud Platform 中的所有资源(包括 BigQuery 表和 Dataflow 作业)都与云项目关联。与 GCP 资源交互时需要指定项目。

异常跟踪表明没有为BigQueryIO.Read转换设置云项目:Caused by: java.lang.NullPointerException: Required parameter projectId must be specified

Dataflow 通过其 API 控制云项目的默认值PipelineOptions。Dataflow 将默认跨其 API 使用该项目,包括BigQueryIO.

通常,我们建议PipelineOptions使用 API 从命令行参数构建PipelineOptionsFactory.fromArgs(String)。在这种情况下,您只需传递--project=YOUR_PROJECT命令行即可。

或者,可以在代码中手动设置,如下所示:

GcpOptions gcpOptions = options.as(GcpOptions.class);
options.setProject("YOUR_PROJECT");
Run Code Online (Sandbox Code Playgroud)

最后,从 Dataflow SDK for Java 1.4.0 版本开始,Dataflow 将默认使用通过gcloud config set project <project>. 您仍然可以通过 覆盖它PipelineOptions,但不需要这样做。即使在版本 1.4.0 之前,这可能在某些场景中有效,但可能并非在所有场景或 Cloud SDK 和 Dataflow SDK 版本组合中都可靠。