Apache Beam：以编程方式创建分区表

Question

我正在编写一个从 Pubsub 读取消息并将其存储到 BigQuery 的云数据流。我想使用分区表（按日期）并且我使用与消息关联的Timestamp 来确定消息应该进入哪个分区。以下是我的代码：

      BigQueryIO.writeTableRows()
        .to(new SerializableFunction<ValueInSingleWindow<TableRow>, TableDestination>() {
            private static final long serialVersionUID = 1L;

            @Override
              public TableDestination apply(ValueInSingleWindow<TableRow> value) {
                log.info("Row value : {}", value.getValue());
                Instant timestamp = value.getTimestamp();
                String partition = DateTimeFormat.forPattern("yyyyMMdd").print(timestamp);
                TableDestination td = new TableDestination(
                    "<project>:<dataset>.<table>" + "$" + partition, null);
                log.info("Table Destination : {}", td);
                return td;
              }
          })            
    .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)         
    .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND) 
    .withSchema(tableSchema);

部署数据流时，我可以在 Stackdriver 中看到日志语句，但是消息没有插入到 BigQuery 表中，并且出现以下错误：

Request failed with code 400, will NOT retry: https://www.googleapis.com/bigquery/v2/projects/<project_id>/datasets/<dataset_id>/tables
severity:  "WARNING"  

所以，它看起来无法创建表，导致插入失败。我是否需要更改数据流定义才能使其正常工作？如果没有，还有其他方法可以通过编程方式创建分区表吗？

我正在使用 Apache Beam 2.0.0。

Answer 1

这是a bug in BigQueryIO，它已在 Beam 2.2 中修复。您可以使用 Beam 的快照版本，或等到 2.2 版最终确定（发布过程目前正在进行中）。