защита паркетных файлов при добавлении

Question

У меня проблема при попытке сделать ETL для большого количества файлов на AWS. Цель - преобразовать файлы JSON в файлы паркета. из-за размера файлов я должен делать это партия за партиями. Допустим, мне нужно сделать это за 15 пакетов, то есть за 15 отдельных прогонов, чтобы иметь возможность преобразовать их все.

Я использую write.mode("append").format("parquet") для записи в файлы паркета в каждом задании pyspark клея, чтобы сделать это.

Моя проблема в том, что если одно задание по какой-то причине не удалось, то я не знаю, что делать - некоторые разделы обновляются, а некоторые нет, некоторые файлы в пакете были обработаны, а некоторые - нет. например, если моя 9-я работа не удалась, я застрял. Я не хочу удалять все файлы паркета, чтобы начать все сначала, но также не хочу просто перезапустить 9-е задание и вызвать дубликаты.

Есть ли способ защитить файлы паркета, добавляя в него новые файлы только в том случае, если все задание выполнено успешно?

Спасибо !!

Answer 1

Исходя из вашего комментария и аналогичного опыта, который у меня был с этой проблемой, я считаю, что это происходит из-за возможной согласованности S3 . Взгляните на Модель согласованности данных Amazon S3 здесь https://docs.aws.amazon.com/AmazonS3/latest/dev/Introduction.html.

Мы обнаружили, что использование секционированного промежуточного s3a коммиттера с режимом разрешения конфликтов replace сделало нашу работу безотказной. Попробуйте следующие параметры для ваших заданий spark:

spark.hadoop.fs.s3a.committer.staging.conflict-mode replace
spark.hadoop.fs.s3a.committer.name partitioned

Также ознакомьтесь с committers здесь: https://hadoop.apache.org/docs/r3.1.1/hadoop-aws/tools/hadoop-aws/committers.html

Надеюсь, это поможет!

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