BigQuery in Dataflow fails to load data from Cloud Storage: JSON object specified for non-record field

Question

I have a Dataflow pipeline running locally on my machine writing to BigQuery. BigQuery in this batch job, requires a temporary location. I have provided one in my Cloud Storage. The relevant parts are:

PipelineOptions options = PipelineOptionsFactory.create();
    options.as(BigQueryOptions.class)
            .setTempLocation("gs://folder/temp");
    Pipeline p = Pipeline.create(options);

....

List<TableFieldSchema> fields = new ArrayList<>();
      fields.add(new TableFieldSchema().setName("uuid").setType("STRING"));
      fields.add(new TableFieldSchema().setName("start_time").setType("TIMESTAMP"));
      fields.add(new TableFieldSchema().setName("end_time").setType("TIMESTAMP"));
      TableSchema schema = new TableSchema().setFields(fields);

session_windowed_items.apply(ParDo.of(new FormatAsTableRowFn()))
      .apply(BigQueryIO.Write
      .withSchema(schema)
      .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
      .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
      .to("myproject:db.table"));

Where for FormatAsTableRowFn I have:

static class FormatAsTableRowFn extends DoFn<KV<String, String>, TableRow>
implements RequiresWindowAccess{  
    @Override
        public void processElement(ProcessContext c) {
          TableRow row = new TableRow()
              .set("uuid", c.element().getKey())
              // include a field for the window timestamp
             .set("start_time", ((IntervalWindow) c.window()).start().toInstant()) //NOTE: I tried both with and without 
             .set("end_time", ((IntervalWindow) c.window()).end().toInstant());   // .toInstant receiving the same error
          c.output(row);
        }
      }

If I print out row.toString() I will get legit timestamps:

{uuid=00:00:00:00:00:00, start_time=2016-09-22T07:34:38.000Z, end_time=2016-09-22T07:39:38.000Z}

When I run this code JAVA says: Failed to create the load job beam_job_XXX

Manually inspecting the temp folder in GCS, the objects look like:

{"mac":"00:00:00:00:00:00","start_time":{"millis":1474529678000,"chronology":{"zone":{"fixed":true,"id":"UTC"}},"zone":{"fixed":true,"id":"UTC"},"afterNow":false,"beforeNow":true,"equalNow":false},"end_time":{"millis":1474529978000,"chronology":{"zone":{"fixed":true,"id":"UTC"}},"zone":{"fixed":true,"id":"UTC"},"afterNow":false,"beforeNow":true,"equalNow":false}}

Looking at the failed job report in BigQuery, the Error says:

JSON object specified for non-record field: start_time (error code: invalid)

This is very strange, because I am pretty sure I said this is a TIMESTAMP, and I am 100% sure my schema in BigQuery conforms with the TableSchema in the SDK. (NOTE: setting the withCreateDisposition...CREATE_IF_NEEDEDyields the same result)

Could someone please tell me how I need to remedy this to get the data inside BigQuery?

Graham Polley Graham Polley · Accepted Answer · 2016-12-28T02:57:33

Don't use Instant objects. Try using milliseconds/seconds.

https://cloud.google.com/bigquery/data-types

A positive number specifies the number of seconds since the epoch

So, something like this should work:

.getMillis() / 1000

BigQuery in Dataflow fails to load data from Cloud Storage: JSON object specified for non-record field

1 Answers