Задание склейки настроено на максимальную емкость 10 узлов, 1 параллельное задание и отсутствие повторных попыток при сбое выдает ошибку «Не удалось удалить ключ: target_folder / _teven», и, согласно stacktrace, проблема заключается в том, что служба S3 начинает блокировать склейку запросы из-за количества запросов: «AmazonS3Exception: пожалуйста, уменьшите частоту запросов.»
Примечание. Проблема не в IAM, поскольку роль IAM, которую использует связующее задание, имеет разрешения на удаление объектов в S3.
Я нашел предложение по этому вопросу на GitHub с предложением уменьшить количество работников: https://github.com/aws-samples/aws-glue-samples/issues/20
"Я добился успеха в сокращении числа работников. "
Однако я не думаю, что 10 - это слишком много работников, и даже хотел бы увеличить количество рабочих до 20, чтобы ускорить ETL.
У кого-нибудь был успех, кто сталкивался с этой проблемой? Как бы я go решил это?
Сокращенная трассировка стека:
py4j.protocol.Py4JJavaError: An error occurred while calling o151.pyWriteDynamicFrame.
: java.io.IOException: Failed to delete key: target_folder/_temporary
at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem.delete(S3NativeFileSystem.java:665)
at com.amazon.ws.emr.hadoop.fs.EmrFileSystem.delete(EmrFileSystem.java:332)
...
Caused by: java.io.IOException: 1 exceptions thrown from 12 batch deletes
at com.amazon.ws.emr.hadoop.fs.s3n.Jets3tNativeFileSystemStore.deleteAll(Jets3tNativeFileSystemStore.java:384)
at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem.doSingleThreadedBatchDelete(S3NativeFileSystem.java:1372)
at com.amazon.ws.emr.hadoop.fs.s3n.S3NativeFileSystem.delete(S3NativeFileSystem.java:663)
...
Caused by: com.amazon.ws.emr.hadoop.fs.shaded.com.amazonaws.services.s3.model.AmazonS3Exception: Please reduce your request rate. (Service: Amazon S3; Status Code: 503; Error Code: SlowDown; Request ID: ...
Часть скрипта Glue ETL python (на всякий случай):
datasource0 = glueContext.create_dynamic_frame.from_catalog(database="database", table_name="table_name", transformation_ctx="datasource0")
... relationalizing, renaming and etc. Transforming from DynamicDataframe to PySpark dataframe and back.
partition_ready = Map.apply(frame=processed_dataframe, f=map_date_partition, transformation_ctx="map_date_partition")
datasink = glueContext.write_dynamic_frame.from_options(frame=partition_ready, connection_type="s3", connection_options={"path": "s3://bucket/target_folder", "partitionKeys": ["year", "month", "day", "hour"]}, format="parquet", transformation_ctx="datasink")
job.commit()
Решено (Вид) , спасибо пользователю ayazabbas
Принял ответ, который помог мне в правильном направлении решения. Одна из вещей, которые я искал, - это как уменьшить множество маленьких файлов на большие порции, и перераспределение делает именно это. Вместо repartition (x) я использовал coalesce (x), где x равно 4 * рабочему количеству связующего задания, чтобы служба Glue могла распределить каждый блок данных для каждого доступного ресурса vCPU. Возможно, имеет смысл иметь x как минимум 2 * 4 * worker_count для учета более медленных и быстрых частей преобразования, если они существуют.
Еще одна вещь, которую я сделал, - это уменьшить количество столбцов, на которые я делил данные перед тем как записать его в S3 с 5 до 4.
Текущий недостаток заключается в том, что я не выяснил, как найти счетчик рабочих в скрипте склейки, который сервис склеивания выделяет для работы, таким образом, число жестко закодировано согласно конфигурации задания (служба Glue выделяет иногда больше узлов, чем настроено).