BigQueryIO.Write Serialization of withJsonSchema

Question

I have a BigQueryIO.Write phase in my beam pipeline that is constructed with a call to .withJsonSchema(String):

inputStream.apply(
    "save-to-bigquery",
    BigQueryIO.<Event>write()
        .withJsonSchema(jsonSchema)
        .to((ValueInSingleWindow<Event> input) ->
                new TableDestination(
                        "table_name$" + PARTITION_SELECTOR.print(input.getValue().getMetadata().getTimestamp()),
                        null)
        )
        .withFormatFunction((ConsumerApiRequest event) ->
                new TableRow()
                        .set("id", event.getMetadata().getUuid())
                        .set("insertId", event.getMetadata().getUuid())
                        .set("account_id", event.getAccountId())
                        ...
                        .set("timestamp", ISODateTimeFormat.dateHourMinuteSecondMillis()
                                .print(event.getMetadata().getTimestamp())))
        .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())
        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
    );

I'm running this via the DataflowRunner and upon execution of this phase I get the following error:

java.lang.IllegalArgumentException: 
        com.google.api.client.json.JsonParser.parseValue(JsonParser.java:889)
        com.google.api.client.json.JsonParser.parse(JsonParser.java:382)
        com.google.api.client.json.JsonParser.parse(JsonParser.java:336)
        com.google.api.client.json.JsonParser.parse(JsonParser.java:312)
        com.google.api.client.json.JsonFactory.fromString(JsonFactory.java:187)
        org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers.fromJsonString(BigQueryHelpers.java:156)
        org.apache.beam.sdk.io.gcp.bigquery.DynamicDestinationsHelpers$ConstantSchemaDestinations.getSchema(DynamicDestinationsHelpers.java:163)
        org.apache.beam.sdk.io.gcp.bigquery.DynamicDestinationsHelpers$ConstantSchemaDestinations.getSchema(DynamicDestinationsHelpers.java:150)
        org.apache.beam.sdk.io.gcp.bigquery.CreateTables$1.processElement(CreateTables.java:103)
Caused by: java.lang.IllegalArgumentException: expected collection or array type but got class com.google.api.services.bigquery.model.TableSchema
        com.google.api.client.repackaged.com.google.common.base.Preconditions.checkArgument(Preconditions.java:148)
        com.google.api.client.util.Preconditions.checkArgument(Preconditions.java:69)
        com.google.api.client.json.JsonParser.parseValue(JsonParser.java:723)
        com.google.api.client.json.JsonParser.parse(JsonParser.java:382)
        com.google.api.client.json.JsonParser.parse(JsonParser.java:336)
        com.google.api.client.json.JsonParser.parse(JsonParser.java:312)
        com.google.api.client.json.JsonFactory.fromString(JsonFactory.java:187)
        org.apache.beam.sdk.io.gcp.bigquery.BigQueryHelpers.fromJsonString(BigQueryHelpers.java:156)
        org.apache.beam.sdk.io.gcp.bigquery.DynamicDestinationsHelpers$ConstantSchemaDestinations.getSchema(DynamicDestinationsHelpers.java:163)
        org.apache.beam.sdk.io.gcp.bigquery.DynamicDestinationsHelpers$ConstantSchemaDestinations.getSchema(DynamicDestinationsHelpers.java:150)
        org.apache.beam.sdk.io.gcp.bigquery.CreateTables$1.processElement(CreateTables.java:103)
        org.apache.beam.sdk.io.gcp.bigquery.CreateTables$1$DoFnInvoker.invokeProcessElement(Unknown Source)
        org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:177)
        org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:141)
        com.google.cloud.dataflow.worker.SimpleParDoFn.processElement(SimpleParDoFn.java:233)
        com.google.cloud.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:48)
        com.google.cloud.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:52)
        com.google.cloud.dataflow.worker.SimpleParDoFn$1.output(SimpleParDoFn.java:183)
        org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue(SimpleDoFnRunner.java:211)
        org.apache.beam.runners.core.SimpleDoFnRunner.access$700(SimpleDoFnRunner.java:66)
        org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:436)
        org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:424)
        org.apache.beam.sdk.io.gcp.bigquery.PrepareWrite$1.processElement(PrepareWrite.java:62)
        org.apache.beam.sdk.io.gcp.bigquery.PrepareWrite$1$DoFnInvoker.invokeProcessElement(Unknown Source)
        .....

It seems that the JSON is properly read at pipeline creation / serialization time but at execution time the deserialized JSON representation is being passed in place of the JSON string. I am generating the JSON string by reading a resource file via the Guava Resources class:

String jsonSchema;
try {
    jsonSchema = Resources.toString(Resources.getResource("path_to_json_schema"), Charsets.UTF_8);
} catch (IOException e) {
    throw new RuntimeException("Failed to load JSON schema", e);
}

How can I fix this serialization issue?

Answer 1

Looking at the code that throws the exception, it appears that this is a JSON parsing failure - your JSON schema is most likely malformed. According to the documentation, it should look something like this:

{
  "fields": [
    {
      "name": string,
      "type": string,
      "mode": string,
      "fields": [
        (TableFieldSchema)
      ],
      "description": string
    }
  ]
}

For example:

{
  "fields": [
    {
      "name": "foo",
      "type": "INTEGER"
    },
    {
      "name": "bar",
      "type": "STRING",
    }
  ]
}

Looking at the code of the JSON parser that is failing, I suspect that you're missing the outer {"fields": ...} and your JSON string only includes the [...].