Apache Beam 的 BigQueryIO (Java)：无法将 TIMESTAMP 字段写入 BigQuery——fastxml.jackson 异常“类型不支持”

Question

我正在尝试 BigQueryIO.write<T>() 使用 Apache Beam 指南 (BigQuery I/O connector) 和 BigQueryIO javadoc 中的代码示例，两者似乎都有一个示例，其中包含一些 Quote 类，该类具有 java Instant 类型的字段（不清楚是java.time.Instant，还是org.joda.time.Instant)，

但是，我从 Jackson 那里得到一个例外，说“不支持类型”，无论如何我都无法让它工作，下面是完整的堆栈。

我不知道在我的管道中哪里可以“添加 com.fasterxml.jackson.datatype:jackson-datatype-joda 模块”。有什么想法吗？

Caused by: com.fasterxml.jackson.databind.exc.InvalidDefinitionException: Joda date/time type `org.joda.time.Instant` not supported by default: add Module "com.fasterxml.jackson.datatype:jackson-datatype-joda" to enable handling (through reference chain: com.google.api.services.bigquery.model.TableRow["timestamp"])
        at com.fasterxml.jackson.databind.exc.InvalidDefinitionException.from(InvalidDefinitionException.java:77)
        at com.fasterxml.jackson.databind.SerializerProvider.reportBadDefinition(SerializerProvider.java:1276)
        at com.fasterxml.jackson.databind.ser.impl.UnsupportedTypeSerializer.serialize(UnsupportedTypeSerializer.java:35)
        at com.fasterxml.jackson.databind.ser.std.MapSerializer.serializeFields(MapSerializer.java:808)
        at com.fasterxml.jackson.databind.ser.std.MapSerializer.serializeWithoutTypeInfo(MapSerializer.java:764)
        at com.fasterxml.jackson.databind.ser.std.MapSerializer.serialize(MapSerializer.java:720)
        at com.fasterxml.jackson.databind.ser.std.MapSerializer.serialize(MapSerializer.java:35)
        at com.fasterxml.jackson.databind.ser.DefaultSerializerProvider._serialize(DefaultSerializerProvider.java:480)
        at com.fasterxml.jackson.databind.ser.DefaultSerializerProvider.serializeValue(DefaultSerializerProvider.java:319)
        at com.fasterxml.jackson.databind.ObjectMapper._writeValueAndClose(ObjectMapper.java:4487)
        at com.fasterxml.jackson.databind.ObjectMapper.writeValueAsString(ObjectMapper.java:3742)
        at org.apache.beam.sdk.io.gcp.bigquery.TableRowJsonCoder.encode(TableRowJsonCoder.java:45)
        at org.apache.beam.sdk.io.gcp.bigquery.TableRowJsonCoder.encode(TableRowJsonCoder.java:32)
        at org.apache.beam.sdk.io.gcp.bigquery.TableRowWriter.write(TableRowWriter.java:42)
        at org.apache.beam.sdk.io.gcp.bigquery.WriteBundlesToFiles.processElement(WriteBundlesToFiles.java:232)

代码是这样的：

 class Quote {
   final Instant timestamp;
   final String exchange;
   final String symbol;
   final double price;

   Quote(Instant timestamp, String exchange, String symbol, double price) {
     // initialize all member variables.
   }
 }

 PCollection<Quote> quotes = ...

 quotes.apply(BigQueryIO
     .<Quote>write()
     .to("my-project:my_dataset.my_table")
     .withSchema(new TableSchema().setFields(
         ImmutableList.of(
           new TableFieldSchema().setName("timestamp").setType("TIMESTAMP"),
           new TableFieldSchema().setName("exchange").setType("STRING"),
           new TableFieldSchema().setName("symbol").setType("STRING"),
           new TableFieldSchema().setName("price").setType("FLOAT"))))
     .withFormatFunction(quote -> new TableRow().set(..set the columns..))
     .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE));

Answer 1

您需要在创建 TableRow 对象时使用 toString() 将 Instant（java.time.Instant 或 org.joda.time.Instant）转换为 String，这样才能工作。

作为参考，文档中的示例 createTableRow 方法涵盖了所有受支持类型的预期格式：https://beam.apache.org/documentation/io/built-in/google-bigquery/#data-types