带有数据流的 Apache Beam：WriteToBigQuery 标志“ignore_unknown_columns”不起作用

Question

我正在使用 Apache Beam（Python SDK 版本 2.37.0）和 Google Dataflow 构建流式传输管道，以将我通过 Pubsub 收到的一些数据写入 BigQuery。

我处理数据并最终得到由这样的字典表示的行：

{'val1': 17.4, 'val2': 40.8, 'timestamp': 1650456507, 'NA_VAL': 'table_name'}

然后我想使用

WriteToBigQuery

将值插入到我的表中。

但是，我的表格只有

val1

、

val2

和

timestamp

列。因此，

NA_VAL

应被忽略。根据我对docs的理解，这应该可以通过设置

ignore_unknown_columns=True

来实现。

但是，在 Dataflow 中运行管道时，我仍然收到错误，并且没有值插入到表中：

There were errors inserting to BigQuery. Will not retry. Errors were [{'index': 0, 'errors': [{'reason': 'invalid', 'location': 'NA_VAL', 'debugInfo': '', 'message': 'no such field: NA_VAL.'}]}]

我尝试过像这样的简单作业配置

rows | beam.io.WriteToBigQuery(
            table='PROJECT:DATASET.TABLE',
            ignore_unknown_columns=True)

以及这些参数

rows | beam.io.WriteToBigQuery(
            table='PROJECT:DATASET.TABLE',
            ignore_unknown_columns=True,
            create_disposition=beam.io.BigQueryDisposition.CREATE_NEVER,
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
            method='STREAMING_INSERTS',
            insert_retry_strategy='RETRY_NEVER')

问题：我是否遗漏了一些阻止管道工作的东西？有人有同样的问题和/或解决方案吗？

Answer 1

不幸的是你被虫子咬了。这被报告为 https://issues.apache.org/jira/browse/BEAM-14039 并由 https://github.com/apache/beam/pull/16999 修复。版本 2.38.0 将包含此修复。该版本的验证今天刚刚结束，因此应该很快就会可用。

Answer 2

对于使用ignore_unknown_columns选项 - 您必须提供架构，如果未提供架构，那么它将尝试将所有列摄取到表中。在流式管道中，您必须提供 schema，但在批处理管道或文件加载方法的情况下，如果您不提供 schema，则ignore_unknown_columns 将不起作用，这很奇怪。

method=beam.io.WriteToBigQuery.Method.FILE_LOADS,
ignore_unknown_columns=True,