无法使用 Cloud Dataflow 将可为空的整数值写入 BigQuery答案

【问题标题】：Unable to write nullable integer values to BigQuery using Cloud Dataflow无法使用 Cloud Dataflow 将可为空的整数值写入 BigQuery
【发布时间】：2016-07-23 06:55:17
【问题描述】：

我正在尝试使用 Cloud Dataflow 写入 BigQuery 表。此 BigQuery 表有一个整数列，该列设置为可为空。对于空值，它会给出以下错误：

无法将值转换为整数。字段：ITM_QT；价值：

但是当我将同一列的数据类型转换为字符串时，它接受的是空值。

那么有没有办法使用 Cloud Dataflow 将空值写入整数列？

如果我将列数据类型更改为字符串，则会出现此错误。

【问题讨论】：

即使对于空时间戳值也存在此问题。所以我面临整数、浮点数和时间戳数据类型的这个问题。
您确定您设置的是 null（实际为 null），而不是字符串值中的“null”吗？

标签： google-bigquery google-cloud-dataflow

【解决方案1】：

不确定您做错了什么，但以下代码可以正常工作，并且确实允许在 BigQuery 中为 Integer 和 Float 数据类型写入 null 值：

public static void main(String[] args) {
        DataflowPipelineOptions options = PipelineOptionsFactory.create().as(DataflowPipelineOptions.class);
        options.setRunner(DirectPipelineRunner.class);
        options.setProject("<project-id>");

        Pipeline pipeline = Pipeline.create(options);

        PCollection<TableRow> results = pipeline.apply("whatever", BigQueryIO.Read.from("<table-spec>")).apply(ParDo.of(new DoFn<TableRow, TableRow>() {
            @Override
            public void processElement(ProcessContext c) throws Exception {
                System.out.println(c.element());
                TableRow row = new TableRow();
                row.set("foo", null); //null FLOAT
                row.set("bar", null); //null INTEGER
                c.output(row);
            }
        }));

        List<TableFieldSchema> fields = new ArrayList<>();
        fields.add(new TableFieldSchema().setName("foo").setType("FLOAT"));
        fields.add(new TableFieldSchema().setName("bar").setType("INTEGER"));
        TableSchema schema = new TableSchema().setFields(fields);

        results.apply(BigQueryIO.Write
                .named("Write")
                .to("<project-id>:<dataset-name>.write_null_numbers_test")
                .withSchema(schema)
                .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
                .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

        pipeline.run();
    }

【讨论】：

嘿，谢谢，您的解决方案确实有效。错误来了，因为我做错了。