Луч разбор JSON - PullRequest
Новая
       4

Луч разбор JSON

0 голосов
/ 31 мая 2018

Я пытаюсь прочитать и проанализировать файл JSON в коде Apache Beam. 

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

Ниже приведен пример JSON, который мне нужно проанализировать testdata из этого файла: myfirst.json 

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

Может ли кто-нибудь подсказать, как проанализировать testdata и получить содержимое из вышеуказанного файла JSON, а затем мне нужно выполнить потоковую передачу данных с помощью Beam?

Ответы [ 2 ]

0 голосов
/ 04 июня 2018

Да, современные библиотеки JSON позволят вам анализировать совершенно произвольный поток JSON и псевдо-JSON в поток объектов, не загружая весь файл в память.

Нет особой необходимостисобрать ваши объекты в одну строку.Действительно, при разработке программного обеспечения для обработки больших данных рекомендуется избегать резервирования огромных объемов памяти для пакетной обработки ваших данных, когда возможна потоковая передача по требованию с использованием простых килобайт памяти.

Взятьпосмотрите короткий учебник по Baeldung: http://www.baeldung.com/jackson-streaming-api

Я включу здесь центральные части кода статьи Baeldung, так как это хорошая практика на случай, если сайт выйдет из строя: 

while (jParser.nextToken() != JsonToken.END_OBJECT) {
    String fieldname = jParser.getCurrentName();
    if ("name".equals(fieldname)) {
        jParser.nextToken();
        parsedName = jParser.getText();
    }

    if ("age".equals(fieldname)) {
        jParser.nextToken();
        parsedAge = jParser.getIntValue();
    }

    if ("address".equals(fieldname)) {
        jParser.nextToken();
        while (jParser.nextToken() != JsonToken.END_ARRAY) {
            addresses.add(jParser.getText());
        }
    }
}

В этом случае анализатор начинается с токена запуска объекта и переходит к анализу этого объекта.В вашем случае вы захотите продолжать цикл до тех пор, пока ваш файл не будет готов, поэтому после того, как вы выйдете из цикла while, вы продолжите двигаться вперед, пока не найдете JsonToken.START_OBJECT, затем создадите новый объект, пройдете черезэта процедура синтаксического анализа, и, в конце концов, передать объект Apache Beam.

0 голосов
/ 31 мая 2018

Прежде всего, я не думаю, что возможно (или, по крайней мере, распространено) обрабатывать «хорошо напечатанный» JSON.Вместо этого данные JSON обычно поступают из 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 у вас будет потоклиний ".Далее, вы можете map этот "поток строк" в "поток JSON", применяя функцию разбора в ParDo: 

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"
...