Nsryjdtyk

For a simple proof-of-concept, I'm trying to window click-data in two minute windows. All I want to do from there is print the per-window count, along with the windows' boundaries to BigQuery. On running my pipeline I keep receiving the following error:

org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.RuntimeException: java.io.IOException: Insert failed: [{"errors":[{"debugInfo":"","location":"windowend","message":"This field is not a record.","reason":"invalid"}],"index":0}]

The pipeline looks like this:

// Creating the pipeline

Pipeline p = Pipeline.create(options);



// Window items

PCollection<TableRow> counts = p.apply("ReadFromPubSub", PubsubIO.readStrings().fromTopic(options.getTopic()))

.apply("AddEventTimestamps", WithTimestamps.of(TotalCountPipeline::ExtractTimeStamp).withAllowedTimestampSkew(Duration.standardDays(10000)))

        .apply("Window", Window.<String>into(

                FixedWindows.of(Duration.standardHours(options.getWindowSize())))

                .triggering(

                        AfterWatermark.pastEndOfWindow()

                                .withLateFirings(AfterPane.elementCountAtLeast(1)))

                .withAllowedLateness(Duration.standardDays(10000))

                .accumulatingFiredPanes())

        .apply("CalculateSum", Combine.globally(Count.<String>combineFn()).withoutDefaults())

        .apply("BigQueryFormat", ParDo.of(new FormatCountsFn()));



// Writing to BigQuery

counts.apply("WriteToBigQuery",BigQueryIO.writeTableRows()

                .to(options.getOutputTable())

                .withSchema(getSchema())

.withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND));



// Execute pipeline

p.run().waitUntilFinish();

I'm guessing it has something to do with the BigQuery formatting function, which is implemented as follows:

static class FormatCountsFn extends DoFn<Long, TableRow> {

    @ProcessElement

    public void processElement(ProcessContext c, BoundedWindow window) {

        TableRow row =

                new TableRow()

                        .set("windowStart", window.maxTimestamp().toDateTime())

                        .set("count", c.element().intValue());

        c.output(row);

    }

}

As inspired by this post. Can anyone shed some light on this? Can't seem to get my head around it.

Apparently the answer this question had nothing to do with beam windowing and was solely related to BigQuery. Writing a DateTime object to a BigQuery row requires a string in the proper yyyy-MM-dd HH:mm:ss format, this in contrast to the DateTime object I was providing.