【发布时间】：2019-11-18 02:57:11
【问题描述】：
我正在创建一个演示管道，以使用我的免费 Google 帐户将 CSV 文件加载到带有 Dataflow 的 BigQuery 中。这就是我所面临的。
当我从 GCS 文件中读取数据并记录数据时，这非常有效。下面是我的示例代码。
这段代码运行正常
DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
options.setProject("project12345");
options.setStagingLocation("gs://mybucket/staging");
options.setRunner(DataflowRunner.class);
DataflowRunner.fromOptions(options);
Pipeline p = Pipeline.create(options);
p.apply(TextIO.read().from("gs://mybucket/charges.csv")).apply(ParDo.of(new DoFn<String, Void>() {
@ProcessElement
public void processElement(ProcessContext c) {
LOG.info(c.element());
}
}));
但是，当我添加一个带有我创建的存储桶路径的临时文件夹位置时，我收到一个错误，下面是我的代码。
LOG.debug("Starting Pipeline");
DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
options.setProject("project12345");
options.setStagingLocation("gs://mybucket/staging");
options.setTempLocation("gs://project12345/temp");
options.setJobName("csvtobq");
options.setRunner(DataflowRunner.class);
DataflowRunner.fromOptions(options);
Pipeline p = Pipeline.create(options);
boolean isStreaming = false;
TableReference tableRef = new TableReference();
tableRef.setProjectId("project12345");
tableRef.setDatasetId("charges_data");
tableRef.setTableId("charges_data_id");
p.apply("Loading Data from GCS", TextIO.read().from("gs://mybucket/charges.csv"))
.apply("Convert to BiqQuery Table Row", ParDo.of(new FormatForBigquery()))
.apply("Write into Data in to Big Query",
BigQueryIO.writeTableRows().to(tableRef).withSchema(FormatForBigquery.getSchema())
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
.withWriteDisposition(isStreaming ? BigQueryIO.Write.WriteDisposition.WRITE_APPEND
: BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE));
p.run().waitUntilFinish();
}
当我运行它时，我收到以下错误
Exception in thread "main" java.lang.IllegalArgumentException: DataflowRunner requires gcpTempLocation, but failed to retrieve a value from PipelineOptions
at org.apache.beam.runners.dataflow.DataflowRunner.fromOptions(DataflowRunner.java:242)
at demobigquery.StarterPipeline.main(StarterPipeline.java:74)
Caused by: java.lang.IllegalArgumentException: Error constructing default value for gcpTempLocation: tempLocation is not a valid GCS path, gs://project12345/temp.
at org.apache.beam.sdk.extensions.gcp.options.GcpOptions$GcpTempLocationFactory.create(GcpOptions.java:247)
at org.apache.beam.sdk.extensions.gcp.options.GcpOptions$GcpTempLocationFactory.create(GcpOptions.java:228)
at org.apache.beam.sdk.options.ProxyInvocationHandler.returnDefaultHelper(ProxyInvocationHandler.java:592)
at org.apache.beam.sdk.options.ProxyInvocationHandler.getDefault(ProxyInvocationHandler.java:533)
at org.apache.beam.sdk.options.ProxyInvocationHandler.invoke(ProxyInvocationHandler.java:155)
at com.sun.proxy.$Proxy15.getGcpTempLocation(Unknown Source)
at org.apache.beam.runners.dataflow.DataflowRunner.fromOptions(DataflowRunner.java:240)
这是身份验证问题吗？因为我通过 Eclipse Dataflow 插件从 GCP 使用 JSON 凭据作为项目所有者。
任何帮助将不胜感激。
【问题讨论】：
-
您的 tempLocation 是有效的 GCS URI 吗？ beam.apache.org/documentation/runners/dataflow/…
-
可能与您的问题重复，但尚不清楚为什么它是与身份验证相关的问题。 stackoverflow.com/questions/43026371/…
-
它是一个有效的 URL，我可以浏览到我指定的存储桶。
标签： java google-cloud-platform google-cloud-storage google-cloud-dataflow dataflow