google-cloud-dataflow - 不可调试的数据流异常

Question

我们在 Dataflow 中创建了一个非常简单的作业。它定期从 BigQuery 读取并存储到 RedisCluster 中。管道代码如下。

    // initialize pipeline
    PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().create();
    Pipeline pipeline = Pipeline.create(options);

    // ... read from bigquery
    final String query = "..."; // actual query snipped
    PCollection<TableRow> bigQueryRes = pipeline.apply("Reading from BigQuery table",
            BigQueryIO.readTableRows().fromQuery(query).usingStandardSql());

    // ... store to Redis Cluster
    bigQueryRes.apply("Storing to RedisCluster",
            ParDo.of(new RedisClusterSinkDoFn()));

    // run pipeline
    pipeline.run();

当我们直接通过命令行直接执行作业时，它执行得很好。

但是当我们从同一个 JAR 生成模板时，使用 launch api 通过 Cloud Function 执行我们发现以下异常:

 java.io.FileNotFoundException: No files matched spec: gs://dataflow-bucket/tmp/BigQueryExtractTemp/0e22ca710a6b45f7aa134cff02b4ae9b/000000000057.avro

    at org.apache.beam.sdk.io.FileSystems.maybeAdjustEmptyMatchResult (FileSystems.java:166)
    at org.apache.beam.sdk.io.FileSystems.match (FileSystems.java:153)
    at org.apache.beam.sdk.io.FileBasedSource.createReader (FileBasedSource.java:332)
    at org.apache.beam.runners.dataflow.worker.WorkerCustomSources$1.iterator (WorkerCustomSources.java:362)
    at org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.runReadLoop (ReadOperation.java:179)
    at org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.start (ReadOperation.java:160)
    at org.apache.beam.runners.dataflow.worker.util.common.worker.MapTaskExecutor.execute (MapTaskExecutor.java:77)
    at org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.executeWork (BatchDataflowWorker.java:395)
    at org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.doWork (BatchDataflowWorker.java:364)
    at org.apache.beam.runners.dataflow.worker.BatchDataflowWorker.getAndPerformWork (BatchDataflowWorker.java:292)

我们不确定是什么导致了此问题，因为 Dataflow 日志和 BigQuery 都没有帮助。我能看出的唯一区别是，当作业成功时 - 有一个文件夹“gs://dataflow-bucket/tmp/BigQueryExtractTemp”，它是使用查询结果的拆分输出创建的。该文件夹不存在，作业失败。

对于从哪里开始调试有什么想法吗？

感谢期待。

Answer 1

我一直在调查此问题，发现 BigQueryIO 中使用的 jobId(用于 BigQuery 作业)是在创建模板时生成的。这意味着，对于同一模板的第二次执行，它将生成 409 错误，表明具有相同 jobId 的 BigQuery 作业已存在。由于尚未为 BigQuery 输入数据创建临时文件，因此导致出现“java.io.FileNotFoundException:没有与规范匹配的文件:”错误。

幸运的是，这个错误已在 Apache Beam 2.x 版本中得到修复。 Google documentation说:

“如果您想要运行从 BigQuery 读取数据的批处理管道，则必须在所有 BigQuery 读取中使用 .withTemplateCompatibility()”

因此，要修复此问题，请将 .withTemplateCompatibility() 添加到您的 BigQueryIO 读取中。

希望对您有帮助!