Как вы читаете файлы JSON в Apache Beam (поток данных) через Python? - PullRequest
Новая
       27

Как вы читаете файлы JSON в Apache Beam (поток данных) через Python?

1 голос
/ 06 мая 2019

Я пытаюсь прочитать файл JSON через apache beam в python и применить к нему некоторые правила качества данных.В настоящее время я использую beam.io.ReadFromText для чтения каждой строки json и использую некоторые функции для изменения данных.Что может быть лучше для чтения данных JSON и их изменения? 

(p
  | 'Getdata' >> beam.io.ReadFromText(input)
  | 'filter_name' >> beam.FlatMap(lambda line: dq_name(line))
  | 'filter_phone' >> beam.FlatMap(lambda line: dq_phone(line))
  | 'filter_zip' >> beam.FlatMap(lambda line: dq_zip(line))
  | 'filter_address' >> beam.FlatMap(lambda line: dq_city(line))
  | 'filter_website' >> beam.FlatMap(lambda line: dq_website(line))
  | 'write' >> beam.io.WriteToText(output_prefix)  )

Примечание. Я довольно новичок в этом, извините, если мой нынешний подход выглядит слишком дурацким.

Ответы [ 2 ]

0 голосов
/ 07 мая 2019

Я думаю, что ваш трубопровод в порядке.Он будет работать параллельно без каких-либо проблем.К вашему сведению, если вы используете FlatMap только для фильтрации элементов, вы также можете использовать Filter.

0 голосов
/ 06 мая 2019

Вы приближаетесь к Apache Beam (Поток данных) с неправильного направления.

Вы пытаетесь прочитать строку и затем применить преобразования к этой строке по одному.

Вместо этого вынужно смотреть на Beam, являющийся параллельным процессором.Вы будете читать все строки ReadFromText(), а затем применять преобразования к каждой строке параллельно.

Посмотрите на функцию beam.ParDo().Это позволит вам создать класс, который сможет обрабатывать каждую строку вашего файла JSON.В этом случае ваш код будет состоять из основных шагов, таких как ReadFromText(), ParDo(MyJsonProcessor()), WriteToText().

Помните, что ваш JSON должен быть JSON с разделителем-новой строкой.http://ndjson.org/

...