Dataflow 中的 DirectPipelineRunner 从本地机器读取到 Google Cloud 存储答案

【问题标题】：DirectPipelineRunner in Dataflow to read from Local machine to Google Cloud storageDataflow 中的 DirectPipelineRunner 从本地机器读取到 Google Cloud 存储
【发布时间】：2018-08-09 20:37:37
【问题描述】：

我尝试运行 Dataflow 管道以从本地计算机（Windows）读取并使用 DirectPipelineRunner 写入 Google 云存储。该作业失败，并出现以下指定 FileNotFoundException 的错误（因此我相信数据流作业无法读取我的位置）。我正在从我的本地计算机运行作业以运行我创建的基于 GCP 的模板。我可以在 GCP Dataflow 仪表板中看到它，但失败并出现以下错误。请帮忙。我还尝试了本地计算机的 IP 或主机名以及本地位置，但遇到了这个 FileNotFoundException？

错误：

java.io.FileNotFoundException: No files matched spec: C:/data/sampleinput.txt
    at org.apache.beam.sdk.io.FileSystems.maybeAdjustEmptyMatchResult(FileSystems.java:172)
    at org.apache.beam.sdk.io.FileSystems.match(FileSystems.java:158)
    at org.apache.beam.sdk.io.FileBasedSource.split(FileBasedSource.java:261)
    at com.google.cloud.dataflow.worker.WorkerCustomSources.splitAndValidate(WorkerCustomSources.java:275)

运行模板的命令：

gcloud dataflow jobs run jobname --gcs-location gs://<somebucketname of template>/<templatename> --parameters inputFilePattern=C:/data/sampleinput.txt,outputLocation=gs://<bucketname>/output/outputfile,runner=DirectPipelineRunner

代码：

PCollection<String>  textData =pipeline.apply("Read Text Data", TextIO.read().from(options.getInputFilePattern()));
    textData.apply("Write Text Data",TextIO.write().to(options.getOutputLocation()));

【问题讨论】：

标签： google-cloud-dataflow apache-beam dataflow

【解决方案1】：

gcloud dataflow jobs run 命令在 Cloud Dataflow 上运行您的作业。这意味着 Dataflow 工作人员将尝试查找 C:/data/sampleinput.txt，显然这些工作人员不存在。

您可以通过将sampleinput.txt 上传到存储桶并将URI gs://<bucketname>/sampleinput.txt 提供为inputFilePattern 来解决此问题。然后，Dataflow 工作人员将能够找到您的输入文件，并且作业应该会成功。

【讨论】：

谢谢。只需使用 DirectRunner，它就可以工作。但我需要进行身份验证才能访问其他 GCP 数据源
是的。您可以使用与运行 Dataflow 作业相同的 GCP 凭据来上传文件。