Dataflow中的DirectPipelineRunner可以从本地计算机读取到Google云端存储

Question

我尝试运行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()));

Answer 1

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

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