Зачем нужны коммитеры S3A: решаем проблемы совместимости Amazon S3 с Hadoop HDFS

В поддержку курса Hadoop для инженеров данных сегодня разберем, в чем проблема безопасной отправки заданий и файлов в облачное хранилище Amazon S3 и как ее решить. Читайте далее, почему AWS S3 не дает гарантий согласованности как HDFS, из-за чего S3Guard не обеспечивает транзакционность и как настроить коммиттеры S3A для Spark 2.4.5 с Hadoop 3.1.

Почему AWS S3 не заменит Hadoop HDFS: объектное Cloud-хранилище vs распределенная файловая система

В отличие от Hadoop HDFS, Amazon S3 – это не файловая система, а объектное хранилище, где данные хранятся в виде объектов в ресурсах, называемых корзинами или бакетами (buckets). Доступ к ним можно получить через API Amazon S3 или с помощью консоли Amazon S3. На практике при перечислении, чтении, обновлении или удалении файлов из AWS S3 возникают проблемы несогласованности, т.к. в этом объектном хранилище многие функции работы с файлами (блокировка, переименование, списки контроля доступа и пр.), реализованы по-другому, чем в Apache Hadoop или отсутствуют вовсе.

Несмотря на то, что интеграция AWS с клиентом Hadoop представляет корзину S3 как файловую систему, bucket по-прежнему является объектным хранилищем и отличается рядом ключевых ограничений [1]:

• операции с каталогами потенциально медленные и неатомарные;
• поддерживаются не все файловые операции, например, отсутствует переименование;
• данные не видны в объектном хранилище, пока не записан весь выходной поток;
• Amazon S3 гарантирует согласованность в конечном счете (eventual consistency) – объекты реплицируются между серверами для обеспечения доступности, но изменения данных требуют времени для распространения на другие реплики, что может привести к несогласованности;
• не поддерживаются разрешения файлов и каталогов HDFS и механизм контрольных списков доступа (ACL, Access Control List);
• пропускная способность между вычислительными кластерами и Amazon S3 ограничена и зависит от нагрузки на сеть и виртуальные машины.

Эти ограничения не устраняет даже собственная файловая система AWS EMR (EMRFS, EMR File System), реализация HDFS от Amazon для чтения и записи обычных файлов из EMR в S3. EMRFS не является open-source проектом и доступна только на EMR, а согласованность отслеживает с помощью таблицы DynamoDB. Метаданные используются для отслеживания всех файловых операций (чтение, запись, обновление и копирование), но в них не хранится фактическое содержимое файлов. Эти метаданные используются для проверки того, соответствуют ли объекты из Amazon S3 ожидаемым, позволяя EMRFS проверять согласованность списка и чтения после записи для новых объектов [2].

Поэтому, хотя AWS S3 можно использовать в качестве источника и хранилища для постоянных данных, оно не способно заменить файловую систему в масштабе кластера, такую как Hadoop HDFS, которая отвечает ключевым требованиям к базовой файловой системе [3]:

• при просмотре содержимого каталога с помощью list() должны быть видны все файлы, которые были в нем созданы, а те, что были удалены – нет;
• переименование каталога – это атомарная транзакция: никакой другой процесс в кластере не может переименовать файл или каталог по тому же пути. Если операция rename() не удалась, данные остаются в исходном месте, а в случае успешного выполнения – оказываются в месте назначения.

Хранилище объектов AWS S3 и клиент файловой системы s3a:// не удовлетворяют этим требованиям:

• Amazon S3 имеет несогласованные списки каталогов, если не включен механизм S3Guard – экспериментальная функция для клиента S3A, который может использовать согласованную базу данных в качестве хранилища метаданных об объектах в корзине S3;
• S3A имитирует переименование путем копирования файлов с последующим удалением оригиналов. Это может привести к отказу в процессе работы, и не гарантирует защиту от попытки одновременного выполнения этой операции другим процессом в кластере.

В результате этого возникают следующие проблемы [3]:

• файлы не будут перечислены в каталоге, поэтому они не могут быть переименованы;
• из-за моментной несогласованности удаленные файлы могут быть обнаружены, что останавливает процесс переименования с ошибкой;
• если переименование не удается, данные остаются в неизвестном состоянии;
• если несколько процессов пытаются зафиксировать работу одновременно, выходной каталог может содержать результаты обоих процессов, что лишает эту операцию статуса эксклюзивности. Хотя S3Guard может обеспечить согласованность листинга, время фиксации по-прежнему пропорционально количеству созданных данных и не справляется со сбоем задачи.
• использование классического FileOutputCommmitter для передачи работы Amazon S3 может привести к потере или повреждению сгенерированных данных. По умолчанию Hadoop использует FileOutputFormatCommitter для управления продвижением файлов, созданных при одной попытке задачи до окончательного вывода запроса. Это делается для обработки сбоев задач и заданий, а также для поддержки спекулятивного выполнения через листинг каталогов и переименование их содержимого с фиксацией задач и заданий.
• существующие алгоритмы протокола фиксации (commit protocol) Apache Hadoop не работает безопасно и быстро с «сырым» хранилищем AWS S3: cписок каталогов может быть непоследовательным, задачи и задания могут не включать всю необходимую работу, а переименования вместо быстрых атомарных операций становятся медленными и могут оборваться в процессе выполнения.

Для решения этих проблем в Hadoop реализованы коммитеры S3A для явной поддержки фиксации заданий в S3 через клиент файловой системы S3A в модуле hadoop-aws.

От магии S3Guard до многокомпонентной загрузки: какие бывают коммиттеры S3A

Начиная с Hadoop 3.1 файловая система S3A включает классы для интеграции с протоколами фиксации заданий Hadoop и Spark. Эти классы и являются коммитерами S3A, которые взаимодействуют с файловой системой S3A для надежной фиксации работы в S3. Они реализуют механизм многокомпонентной загрузки (MultiPart Upload, MPU) в S3, позволяя клиенту записывать данные в нескольких запросах HTTP POST. При этом для завершения загрузки выполняется только операция записи с последним POST, который состоит из короткого списка тегов загруженных блоков. Этот механизм многокомпонентной загрузки автоматически используется при записи больших объемов данных в S3 и реализован в выходном потоке S3A. Коммиттеры S3A явно используют MPU-механизм [3]:

• отдельные задачи в задании записывают свои данные в S3 как операции POST в рамках многокомпонентных загрузок, но не выполняют окончательный POST для завершения загрузки;
• многокомпонентные загрузки завершаются (фиксируются) в процессе фиксации задания.

Различают 2 типа коммиттеров [1]:

• промежуточные (Staging) от Netflix, которые бывают двух видов: каталоговые (Directory Committer) и партиционированные (Partitioned Committer);
• магические (Magic) от сообщества Apache Hadoop с использованием S3Guard для обеспечения согласованности данных.

Кратко рассмотрим, как работает каждый вид коммитера:

• Directory Committer буферизует рабочие данные на локальный диск, используя Hadoop HDFS для передачи информации о фиксации задач коммиттеру заданий и разрешая конфликты во всем дереве каталогов назначения;
• Partitioned Committer идентичен коммиттеру каталога, но конфликт управляется для каждого раздела отдельно. Это позволяет использовать его для локального обновления существующих наборов данных, что отлично подходит только для Apache Spark;
• Magic Committer записывает данные непосредственно в S3, «волшебным образом» перенаправляя их в конечный пункт назначения. Конфликт управляется через дерево каталогов с помощью согласованного хранилища объектов S3 и механизма S3Guard.

Чтобы настроить каталоговые коммиттеры S3A для Spark 2.4.5 с Hadoop 3.1, можно использовать следующую опцию [4]:

“spark.hadoop.mapreduce.outputcommitter.factory.scheme.s3a”: “org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory”

“spark.hadoop.fs.s3a.committer.name”: “directory”

“spark.hadoop.fs.s3a.committer.staging.conflict-mode”: “append”

Таким образом, коммитеры S3A решают проблему несогласованности данных и файловых операций при совместном использовании Apache Hadoop и Spark с объектным хранилищем AWS S3. Однако, при этом надо учитывать некоторые особенности интеграции Spark-приложения с объектным хранилищем данных. В следующей статье мы продолжим разговор про Apache Spark на AWS и рассмотрим, как DBT облегчает работу дата-инженера в этом случае. А как получить данные из облачного объектного хранилища AWS S3 с помощью заданий Hive и Spark, включая настройку конфигурационных xml-файлов, мы описываем здесь.

Узнайте больше про администрирование кластеров и разработку распределенных приложений аналитики больших данных с Apache Hadoop и Spark на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:

• Основы Hadoop
• Hadoop для инженеров данных
• Основы Apache Spark для разработчиков

1. https://github.com/aws-samples/eks-spark-benchmark/blob/master/performance/s3.md
2. https://docs.aws.amazon.com/emr/latest/ManagementGuide/emrfs-metadata.html
3. https://hadoop.apache.org/docs/r3.2.1/hadoop-aws/tools/hadoop-aws/committers.html
4. https://aws.amazon.com/ru/blog/containers/optimizing-spark-performance-on-kubernetes/