使用ValueProvider从数据流模板读取BigQuery时出现java异常

Question

我试图创建一个模板来读取BigQuery，不幸的是，我在构建模板时遇到了一个异常

An exception occured while executing the Java class. Cannot call validate if table is dynamically set.

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

Note: If you want to run a batch pipeline that reads from BigQuery, you must use .withTemplateCompatibility() on all BigQuery reads.

下面是我的代码片段：

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。请注意其中提到的解释

Note that a table or query check can fail if the table or dataset are created by earlier stages of the pipeline or if a query depends on earlier stages of a pipeline.

要克服这个问题，请尝试在BigQueryIO中添加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());