使用 ValueProvider 从 Dataflow 模板读取 BigQuery 时出现异常

Question

我正在尝试创建一个从 BigQuery 读取的模板，不幸的是，我在尝试构建该模板时遇到了异常。

执行 Java 类时发生异常。如果表是动态设置的，则无法调用 validate。

阅读the documentation，似乎在从批处理模板中读取BigQuery时需要调用一个特殊的函数：

注意：如果您想运行从 BigQuery 读取数据的批处理流水线， 您必须在所有 BigQuery 读取中使用 .withTemplateCompatibility()。

所以，这是我的代码 sn-p ：

PCollection<Discount> discountFromBigQuery = p.apply("Parse Discounts from BigQuery", BigQueryIO.read((SerializableFunction<SchemaAndRecord, Discount>) record -> {
        GenericRecord row = record.getRecord();
        return new Discount(row);
    }).withTemplateCompatibility().from(options.getBigQueryDiscountPath()).withCoder(SerializableCoder.of(Discount.class)));

显然，options.getBigQueryDiscountPath() 是 ValueProvider<String>

那么，我怎样才能摆脱这个错误并模板化 BigQuery 阅读部分？

这是我使用的 maven 依赖项：

<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-core</artifactId>
    <version>2.8.0</version>
</dependency>
<dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
    <version>2.8.0</version>
</dependency>
<dependency>
    <groupId>com.google.cloud.dataflow</groupId>
    <artifactId>google-cloud-dataflow-java-sdk-all</artifactId>
    <version>2.5.0</version>
</dependency>

Answer 1

我相信您面临的错误定义为here。请注意解释，提到

请注意，如果创建表或数据集，则表或查询检查可能会失败 通过管道的早期阶段，或者如果查询依赖于早期 管道的各个阶段。

要克服这个问题，请尝试在您的 BigQueryIO.read 调用中添加 withoutValidation method。

Answer 2

顺便说一句，需要在链的末尾添加 withoutValidation() ，如下所示。

    // queryString is of type ValueProvider<String>
    PCollection<TableRow> rowsFromBigQuery = pipeline.apply(
                BigQueryIO.readTableRows()
                        .fromQuery(queryString)
                        .usingStandardSql()
                        .withMethod(options.getReadMethod())
                        .withoutValidation());