Запись в динамическое место назначения в облачное хранилище в потоке данных в Python

Я пытался читать из большого файла в облачном хранилище и сегментировать их в соответствии с заданным полем.

Я планирую прочитать | Map(лямбда x: (x[ключевое поле], x)) | группа по ключу | Запись в файл с именем ключевого поля.

Однако я не смог найти способ динамической записи в облачное хранилище. Поддерживается ли этот функционал?

Спасибо, Ицин.


apache-beam python-2.7 google-cloud-storage google-cloud-dataflow apache-beam-io
person yiqing_hua    schedule 15.02.2018    source источник

Ответы (2)


arrow_upward
1
arrow_downward

Да, вы можете использовать FileSystemsAPI для создания файлов.

person jkff    schedule 16.02.2018
comment
Благодарю вас! Я использовал дескриптор FileSystems.create внутри ParDo для записи сгруппированных результатов. Однако похоже, что GroupByKey будет ждать, пока все данные будут прочитаны, а затем начнет запись в один файл. Итак, у меня есть два дополнительных вопроса: 1) Могу ли я использовать подстановочные знаки с помощью FileSystems API? 2) Есть ли способ, чтобы GroupBy не приходилось ждать всех данных, иначе может возникнуть проблема с памятью. Спасибо еще раз! - person yiqing_hua; 22.02.2018

arrow_upward
1
arrow_downward

Экспериментальная запись была добавлена ​​в SDK Beam python в версии 2.14.0, beam.io.fileio.WriteToFiles:

my_pcollection | beam.io.fileio.WriteToFiles(
      path='/my/file/path',
      destination=lambda record: 'avro' if record['type'] == 'A' else 'csv',
      sink=lambda dest: AvroSink() if dest == 'avro' else CsvSink(),
      file_naming=beam.io.fileio.destination_prefix_naming())

который можно использовать для записи в разные файлы для каждой записи.

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

Больше документации здесь:

https://beam.apache.org/releases/pydoc/2.14.0/apache_beam.io.fileio.html#dynamic-destinations

И проблема JIRA здесь:

https://issues.apache.org/jira/browse/BEAM-2857

person anrope    schedule 18.08.2019