É possível ler arquivos JSON não aninhados no Cloud Storage with Dataflow via:

p.apply("read logfiles", TextIO.Read.from("gs://bucket/*").withCoder(TableRowJsonCoder.of()));

Se eu quiser gravar esses logs com filtragem mínima no BigQuery, posso fazer isso usando um DoFn como este:

private static class Formatter extends DoFn<TableRow,TableRow> {

        @Override
        public void processElement(ProcessContext c) throws Exception {

            // .clone() since input is immutable
            TableRow output = c.element().clone();

            // remove misleading timestamp field
            output.remove("@timestamp");

            // set timestamp field by using the element's timestamp
            output.set("timestamp", c.timestamp().toString());

            c.output(output);
        }
    }
}

No entanto, não sei como acessar campos aninhados no arquivo JSON dessa maneira.

Se o TableRow contiver umRECORD nomeador, é possível acessar suas chaves / valores sem serialização / desserialização?Se eu precisar serializar / desserializar-me com oJackson biblioteca, faz mais sentido usar um padrãoCoder doTextIO.Read ao invés deTableRowJsonCoder, para recuperar parte do desempenho que perdi dessa maneira?

EDITAR

Os arquivos são delimitados por nova linha e têm a seguinte aparência:

{"@timestamp":"2015-x", "message":"bla", "r":{"analyzed":"blub", "query": {"where":"9999"}}}
{"@timestamp":"2015-x", "message":"blub", "r":{"analyzed":"bla", "query": {"where":"1111"}}}