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

【问题标题】：Reading nested JSON in Google Dataflow / Apache Beam在 Google Dataflow / Apache Beam 中读取嵌套 JSON
【发布时间】：2017-02-01 16:00:34
【问题描述】：

可以通过以下方式使用 Dataflow 读取 Cloud 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 文件中的嵌套字段。

如果 TableRow 包含一个名为 r 的 RECORD，是否可以在不进一步序列化/反序列化的情况下访问其键/值？
如果我需要使用Jackson 库对自己进行序列化/反序列化，使用TextIO.Read 的标准Coder 而不是TableRowJsonCoder 是否更有意义，以获得一些我之前的性能？这么松？

编辑

文件以换行符分隔，如下所示：

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

【问题讨论】：

文件的格式如何？它们是换行符分隔，还是换行符可能出现在嵌套记录之一中？
文件是换行符分隔的，我不希望嵌套记录之一中有任何换行符。我编辑了我的问题以包含一个示例。

标签： google-cloud-dataflow apache-beam

【解决方案1】：

您最好的选择可能是按照您在 #2 中描述的操作并直接使用 Jackson。最有意义的是让 TextIO 读取执行它的构建目的——使用字符串编码器从文件中读取行——然后使用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);
    }
  });

请注意，您也可以使用多个 ParDo 来执行此操作。

【讨论】：

我最后用多个ParDos解决了，谢谢。