java - 光束json解析

Question

我正在尝试读取和解析 Apache Beam 代码中的 JSON 文件。

PipelineOptions options = PipelineOptionsFactory.create();
options.setRunner(SparkRunner.class);

Pipeline p = Pipeline.create(options);

PCollection<String> lines = p.apply("ReadMyFile", TextIO.read().from("/Users/xyz/eclipse-workspace/beam-project/myfirst.json"));
System.out.println("lines: " + lines);

下面是我需要从中解析 testdata 的示例 JSON:我的第一个.json

{  
   “testdata":{  
      “siteOwner”:”xxx”,
      “siteInfo”:{  
         “siteID”:”id_member",
         "siteplatform”:”web”, 
         "siteType”:”soap”,
         "siteURL”:”www”
      }
   }
}

有人可以指导如何解析 testdata 并从上述 JSON 文件中获取内容，然后我需要使用 Beam 流式传输数据吗？

Answer 1

首先，我认为处理“打印精美”的 JSON 是不可能的(或者至少是不常见的)。相反，JSON 数据通常是从 newline-delimited JSON 中获取的，因此您的输入文件应如下所示:

{"testdata":{"siteOwner":"xxx","siteInfo":{"siteID":"id_member","siteplatform":"web","siteType":"soap","siteURL":"www,}}}
{"testdata":{"siteOwner":"yyy","siteInfo":{"siteID":"id_member2","siteplatform":"web","siteType":"soap","siteURL":"www,}}}

在那之后，您的代码在 lines 中，您就有了“一行流”。接下来，您可以通过在 ParDo 中应用解析函数，将此“行流”映射为“JSON 流”:

static class ParseJsonFn extends DoFn<String, Json> {

  @ProcessElement
  public void processElement(ProcessContext c) {
    // element here is your line, you can whatever you want, parse, print, etc
    // this function will be simply applied to all elements in your stream
    c.output(parseJson(c.element()))
  }
}

PCollection<Json> jsons = lines.apply(ParDo.of(new ParseJsonFn()))  // now you have a "stream of JSONs"