Apache Beam - чтение JSON и Stream

Question

Я пишу код луча Apache, в котором мне нужно прочитать файл JSON, который был помещен в папку проекта, и прочитать данные и передать их в поток.

Это пример кода для чтения 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-prototype/test.json"));
System.out.println("lines: " + lines);

или я должен использовать,

p.apply(FileIO.match().filepattern("/Users/xyz/eclipse-workspace/beam-prototype/test.json"))

Мне просто нужно прочитать ниже файл json.Прочитайте полный файл testdata из этого файла, а затем выполните его потоковую передачу.

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

Приведенный выше код не читает файл json, он печатает как

lines: ReadMyFile/Read.out [PCollection]

, не могли бы выназовите мне образец?

Answer 1

Это пример кода для чтения JSON.Это правильный способ сделать это?

Чтобы быстро ответить на ваш вопрос, да.Ваш пример кода является правильным способом чтения файла, содержащего JSON, где каждая строка файла содержит один элемент JSON.Входное преобразование TextIO читает файл построчно, поэтому, если один элемент JSON занимает несколько строк, он не будет разбираться.

Второй пример кода имеет тот же эффект.

Приведенный выше код не читает файл json, он печатает как

Ожидаемый результат печати.Переменная lines на самом деле не содержит строк JSON в файле.lines - это PCollection из String с;он просто представляет состояние конвейера после применения преобразования.Доступ к элементам в конвейере может быть сделан путем применения последующих преобразований.Фактическая строка JSON может быть доступна при реализации преобразования.