在Google Dataflow/Apache Beam中读取嵌套的JSON

Question

可以通过以下方式在Data Storage上读取未存储的JSON文件:

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

如果我只想用BigQuery编写那些带有最小过滤的日志,我可以通过使用像这样的DoFn来实现:

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);
        }
    }
}

但是,我不知道如何以这种方式访问​​JSON文件中的嵌套字段.

1. 如果TableRow包含一个RECORDnamed r,是否可以访问其键/值而无需进一步序列化/反序列化？
2. 如果我需要序列化/反序列化自己与Jackson库,它让使用标准更有意义Coder的TextIO.Read替代TableRowJsonCoder,从而获得一些我失去这样的演出回来的？

编辑

文件是换行符的新行,看起来像这样:

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

Answer 1

你最好的选择可能是做你在#2中所描述的并直接使用杰克逊.让TextIO读取它所构建的内容是最有意义的 - 使用字符串编码器从文件中读取行 - 然后使用a DoFn来实际解析元素.类似于以下内容:

PCollection<String> lines = pipeline
  .apply(TextIO.from("gs://bucket/..."));
PCollection<TableRow> objects = lines
  .apply(ParDo.of(new DoFn<String, TableRow>() {
    @Override
    public void processElement(ProcessContext c) {
      String json = c.element();
      SomeObject object = /* parse json using Jackson, etc. */;
      TableRow row = /* create a table row from object */;
      c.output(row);
    }
  });

请注意,您也可以使用多个ParDos执行此操作.