Я новичок в искре и у меня есть некоторые фундаментальные сомнения.Я работаю над приложением pyspark.Предполагается обработать 500 тыс. Предметов.Текущая реализация неэффективна и требует вечного завершения.Я кратко объясню задачи.
Приложение обрабатывает каталог S3.Предполагается обработать все файлы, представленные в s3://some-bucket/input-data/.Структура каталогов S3 выглядит следующим образом:
s3://some-bucket/input-data/item/Q12/sales.csv
s3://some-bucket/input-data/item/Q13/sales.csv
s3://some-bucket/input-data/item/Q14/sales.csv
Файлы csv не имеют столбца идентификатора элемента.Имя каталога - это идентификатор элемента, например Q11, Q12 и т. Д.
Для приложения определен udf, который загружает данные с помощью boto3, обрабатывает их и затем выгружает данные в S3в структуре каталогов вот так:
s3://some-bucket/output-data/item/Q12/profit.csv
s3://some-bucket/output-data/item/Q13/profit.csv
s3://some-bucket/output-data/item/Q14/profit.csv
Выполнение 500K API-вызова S3 для данных, мне не кажется правильным.Я запускаю приложение spark на EMR. Должен ли я загрузить все данные в качестве шага начальной загрузки?
Может ли S3DistCp (s3-dist-cp) решить проблему, загрузив все данные вHDFS и более поздние рабочие / узлы могут получить к ним доступ.Советы по использованию s3-dist-cp были бы очень полезны.