HDFS Erasure Coding

Цель

Репликация всегда дорогостоящая – схема репликации по умолчанию (3x) в HDFS имеет 200% накладных расходов в области хранения и других ресурсах (например, пропускная способность сети). Однако для теплых и холодных наборов данных с относительно низким уровнем операций ввода-вывода дополнительные реплики блоков редко доступны во время обычных операций, но все равно потребляют тот же объем ресурсов, что и первая реплика.

Поэтому естественным улучшением является использование Erasure Coding (EC) вместо репликации, что обеспечивает тот же уровень отказоустойчивости при гораздо меньшем объеме памяти. В типичных настройках Erasure Coding накладные расходы на хранение не превышают 50%. Коэффициент репликации файла EC не имеет смысла, он всегда равен 1 и не может быть изменен с помощью -setrep команды.

Background

В системах хранения наиболее заметным использованием Erasure Coding является избыточный массив недорогих дисков (RAID). RAID реализует EC посредством чередования, которое делит логически последовательные данные (например, файл) на более мелкие единицы (например, бит, байт или блок) и сохраняет последовательные единицы на разных дисках. Далее данная единица распределения чередования называется чередующейся ячейкой (или просто ячейкой). Для каждой полосы исходных ячеек данных вычисляется и сохраняется определенное количество ячеек четности – процесс, который называется кодированием. Ошибка в любой чередующейся ячейке может быть исправлена путем вычисления декодирования на основе сохранившихся данных и четности ячеек.

Интеграция Erasure Coding с HDFS может повысить эффективность хранилища, обеспечивая при этом такую же долговечность данных, что и традиционные развертывания HDFS на основе репликации. Например, -реплицированный файл с 6 блоками будет занимать 6 * 3 = 18 блоков дискового пространства. Но при развертывании EC (6 данных, 3 четности) он будет занимать только 9 блоков дискового пространства.

Архитектура

В контексте Erasure Coding чередование имеет несколько важных преимуществ. Во-первых, позволяется онлайн запись данных непосредственно в формате EC, избегая фазы преобразования и немедленно экономя место для хранения. Это также повышает производительность последовательного ввода-вывода за счет параллельного использования нескольких дисков, что особенно желательно в кластерах с высокопроизводительными сетями. Во-вторых, небольшие файлы естественным образом распределяются на несколько узлов данных DataNodes и устраняется необходимость объединения нескольких файлов в одну группу кодирования. Это значительно упрощает операции с файлами, такие как удаление, quota reporting и миграция между объединенными пространствами имен Namespaces.

В типичных кластерах HDFS небольшие файлы могут занимать более 3/4 общего объема памяти. Чтобы лучше поддерживать небольшие файлы, на этом первом этапе работы HDFS поддерживает EC с чередованием. В будущем HDFS также будет поддерживать смежную компоновку EC.

Расширения NameNode – чередующиеся файлы HDFS, логически состоящие из групп блоков, каждая из которых содержит определенное количество внутренних блоков. Чтобы уменьшить потребление памяти NameNode от этих дополнительных блоков, введен новый иерархический протокол именования блоков. Идентификатор группы блоков может быть выведен из идентификатора любого из ее внутренних блоков. Это позволяет управлять на уровне группы блоков, а не на уровне одного блока.

Клиентские расширения – клиентские пути чтения и записи улучшены для параллельной работы с несколькими внутренними блоками в группе блоков. На пути вывода/записи DFSStripedOutputStream управляет набором потоков данных, по одному для каждого DataNode, хранящего внутренний блок в текущей группе блоков. Стримеры в основном работают асинхронно. Координатор отвечает за операции над всей группой блоков, включая завершение текущей группы блоков, выделение новой группы блоков и так далее. На пути ввода/чтения DFSStripedInputStream преобразует запрошенный логический байтовый диапазон данных в виде диапазонов во внутренние блоки, хранящиеся в DataNodes. Затем параллельно выдаются запросы на чтение. А при сбоях выдаются дополнительные запросы на чтение для декодирования.

Расширения DataNode – DataNode запускает дополнительную задачу ErasureCodingWorker (ECWorker) для фонового восстановления сбойных блоков Erasure Coded. Сбойные блоки EC обнаруживаются NameNode, который затем выбирает DataNode для выполнения работы по восстановлению. Задача восстановления передается как ответ на heartbeat-сообщение. Этот процесс аналогичен тому, как реплицированные блоки повторно реплицируются при сбое. Реконструкция выполняет три ключевые задачи:

  • Чтение данных из исходных узлов: входные данные считываются параллельно из исходных узлов с помощью выделенного пула потоков. Основываясь на политике EC, он планирует запросы на чтение для всех исходных целей и считывает только минимальное количество входных блоков для восстановления;
  • Декодирование данных и генерирование выходных данных: новые данные и блоки четности декодируются из входных данных. Все недостающие данные и блоки четности декодируются вместе;
  • Передача сгенерированных блоков данных на целевые узлы: после завершения декодирования восстановленные блоки передаются на целевые DataNodes.

Политики Erasure Coding – файлам и каталогам в кластере HDFS разрешается использование разных политик репликации и кодирования для обеспечения разнородных рабочих нагрузок. Политика Erasure Coding инкапсулирует способ кодирования/декодирования файла. Каждая политика определяется следующей информацией:

  • Схема EC – включает в себя количество блоков данных и четности в группе EC (например, 6+3), а также codec-алгоритм (например, Reed-Solomon, XOR);
  • Размер чередующейся ячейки – определяет степень детализации операций чтения и записи с чередованием, включая размеры буфера и работу кодирования.

Политики называются codec-num data blocks-num parity blocks-cell size. В настоящее время поддерживаются пять встроенных политик: RS-3-2-1024k, RS-6-3-1024k, RS-10-4-1024k, RS-LEGACY-6-3-1024k, XOR-2-1- 1024k.

Схема по умолчанию REPLICATION также поддерживается. Ее можно установить только для каталога, чтобы заставить каталог принимать схему репликации 3x, а не наследовать политику erasure coding родительского каталога верхнего уровня. Политика позволяет чередовать каталог схемы репликации 3x с каталогом erasure coding.

Политика REPLICATION всегда включена. Из всех политик EC по умолчанию включена RS(6,3).

Подобно политикам хранения HDFS, политики erasure coding устанавливаются на каталог. При создании файла он наследует политику EC своего ближайшего каталога-родителя.

Политики EC уровня каталога влияют только на новые файлы, созданные в этом каталоге. Как только файл создан, его политику можно запросить, но не изменить. Если erasure coding файл переименовывается в каталог с другой политикой EC, файл принимает политику нового каталога EC. Преобразование файла в другую политику ЕС требует перезаписи его данных; поэтому рекомендуется копировать файл (например, через distcp), а не переименовывать его.

Arenadata позволяет пользователям определять свои собственные политики EC с помощью XML-файла, который должен состоять из следующих трех частей:

  • layoutversion: указывает версию формата XML-файла политики ЕС;
  • schemas: включает в себя все пользовательские схемы EC;
  • policies: включает в себя все пользовательские политики EC, и каждая политика включает в себя идентификатор схемы и размер чередующейся ячейки (cellsize).

Пример XML-файла политики ЕС с именем user_ec_policies.xml.template находится в каталоге Hadoop conf.

Intel ISA-L расшифровывается как Intel Intelligent Storage Acceleration Library – это набор оптимизированных низкоуровневых функций с открытым исходным кодом, предназначенных для приложений хранения данных. Библиотека включает в себя быстрые блочные erasure codes типа Reed-Solomon, оптимизированные для наборов команд Intel AVX и AVX2. HDFS erasure coding может использовать ISA-L для ускорения вычислений кодирования и декодирования. ISA-L поддерживает большинство основных операционных систем, включая Linux и Windows. ISA-L не включена по умолчанию.

Развертывание

Конфигурация кластера и оборудования

Erasure coding предъявляет к кластеру дополнительные требования с точки зрения процессора и сети.

Работа по кодированию и декодированию требует дополнительных ресурсов ЦП как для клиентов HDFS, так и для узлов DataNodes.

Для Erasure coding требуется как минимум столько же DataNodes в кластере, сколько сконфигурировано блоков файловой системы EC. Для ЕС политики RS (6,3) это означает минимум 9 узлов DataNodes.

Файлы erasure coding распределяются по стойкам с целью обеспечения ее отказоустойчивости. Это означает, что при чтении и записи чередующихся файлов большинство операций выполняется вне стойки. Таким образом, пропускная способность bisection-сети очень важна.

Для отказоустойчивости стойки также важно иметь достаточное количество стоек, чтобы в среднем каждая стойка содержала количество блоков не большее, чем количество блоков четности EC. Формула для расчета получается: (блоки данных + блоки четности) / блоки четности с округлением в большую сторону. Для политики ЕС RS (6,3) это означает минимум 3 стойки, рассчитанные по формуле (6 + 3) / 3 = 3, но в идеале должно быть 9 или более для обработки запланированных и незапланированных отключений. Для кластеров с меньшим количеством стоек, чем число ячеек четности, HDFS не может поддерживать отказоустойчивость стойки, но при этом все равно пытается распределить чередующийся файл по нескольким узлам для сохранения отказоустойчивости на уровне узла. По этой причине рекомендуется устанавливать стойки с одинаковым количеством узлов DataNodes.

Ключи конфигурации

По умолчанию все встроенные политики erasure coding отключены, за исключением политики, определенной в dfs.namenode.ec.system.default.policy. Администратор кластера может включить набор политик с помощью команды hdfs ec [-enablePolicy -policy <policyName>] в зависимости от размера кластера и требуемых свойств отказоустойчивости. Например, для кластера с 9 стойками такая политика, как RS-10-4-1024k, не сохранит отказоустойчивость на уровне стойки, и RS-6-3-1024k или RS-3-2-1024k могут быть более подходящими. Если администратор заботится об отказоустойчивости только на уровне узла, политика RS-10-4-1024k будет по-прежнему уместной, если в кластере есть по крайней мере 14 DataNodes.

Системная политика ЕС по умолчанию может быть настроена через конфигурацию dfs.namenode.ec.system.default.policy. В этой конфигурации политика EC по умолчанию будет использоваться, когда имя политики не передается в качестве аргумента в команде -setPolicy.

По умолчанию dfs.namenode.ec.system.default.policyRS-6-3-1024k.

Реализация codec для Reed-Solomon и XOR может быть настроена с помощью следующих ключей конфигурации клиента и DataNode: + io.erasurecode.codec.rs.rawcoders для RS codec по умолчанию; + io.erasurecode.codec.rs-legacy.rawcoders для предыдущих версий RS codec; + io.erasurecode.codec.xor.rawcoders для XOR codec.

Пользователь также может настроить самостоятельный кодек с помощью ключа конфигурации, например: io.erasurecode.codec.self-defined-codec.rawcoders. Значения для этого ключа являются списками имен кодеров с резервным механизмом. Эти фабрики кодеков загружаются в заданном значениями конфигурации порядке до тех пор, пока кодек не будет загружен успешно. Конфигурация кодека RS и XOR по умолчанию предпочитает нативную реализацию по сравнению с чистой Java. Реализация нативного кодека RS-LEGACY отсутствует, поэтому по умолчанию используется только реализация Java. Все перечисленные кодеки имеют реализации на чистой Java. Для стандартного кодека RS и кодека XOR существует также собственная реализация, использующая библиотеку Intel ISA-L для повышения производительности кодека. Реализация по умолчанию для RS Legacy – это чистая Java, а реализации по умолчанию для RS и XOR по умолчанию – это собственные реализации, использующие библиотеку Intel ISA-L.

Работы по восстановлению Erasure coding background для узлов DataNodes можно настроить с помощью следующих параметров конфигурации:

  • dfs.datanode.ec.reconstruction.stripedread.timeout.millis – тайм-аут для striped reads. Значение по умолчанию 5000 мс;
  • dfs.datanode.ec.reconstruction.stripedread.buffer.size – размер буфера для сервиса чтения. Значение по умолчанию 64 КБ;
  • dfs.datanode.ec.reconstruction.threads – количество потоков, используемых Datanode для восстановления background. Значение по умолчанию 8 потоков;
  • dfs.datanode.ec.reconstruction.xmits.weight – относительный вес xmits, используемый задачей EC background, по сравнению с восстановлением реплицированного блока. Значение по умолчанию 0,5. Для того, чтобы отключить вычисление весов для задач восстановления EC, необходимо установить значение 0, тогда для задачи EC всегда определен 1 xmits. Xmits задачи восстановления erasure coding вычисляется как максимальное значение между числом потоков чтения и числом потоков записи. Например, если задаче восстановления EC нужно прочитать с 6 узлов и записать на 2 узла, она имеет xmits равный max (6, 2)*0,5 = 3. Задача восстановления для реплицируемого файла всегда считается как 1 xmit. NameNode использует dfs.namenode.replication.max-streams за вычетом общего значения xmitsInProgress для DataNode, который объединяет xmits из реплицированного файла и файлов EC, чтобы запланировать задачи восстановления для этого DataNode.

Включение Intel ISA-L

Собственная реализация стандартного кодека RS в HDFS использует библиотеку Intel ISA-L в целях улучшения вычислений кодирования и декодирования. Чтобы включить и использовать Intel ISA-L, необходимо выполнить три шага:

  1. Сборка библиотеки ISA-L. Подробная информация приведена на официальной странице https://github.com/01org/isa-l/.
  2. Сборка Hadoop с поддержкой ISA-L.
  3. Копирование содержимого каталога isal.lib в конечный файл tar с помощью -Dbundle.isal. Развернуть Hadoop с помощью файла tar. Убедиться, что ISA-L доступна на HDFS клиентах и DataNodes.

Чтобы убедиться, что ISA-L правильно определена в Hadoop, необходимо выполнить команду hadoop checknative.

Команды администрирования

HDFS предоставляет подкоманду ec для выполнения административных команд, связанных с erasure coding:

hdfs ec [generic options]
  [-setPolicy -path <path> [-policy <policyName>] [-replicate]]
  [-getPolicy -path <path>]
  [-unsetPolicy -path <path>]
  [-listPolicies]
  [-addPolicies -policyFile <file>]
  [-listCodecs]
  [-enablePolicy -policy <policyName>]
  [-disablePolicy -policy <policyName>]
  [-help [cmd ...]]

Подробнее о каждой команде:

  • [-setPolicy -path <path> [-policy <policyName>] [-replicate]] – устанавливает политику erasure coding для каталога по указанному пути:
    • path – каталог в HDFS. Обязательный параметр. Установка политики влияет только на вновь созданные файлы и не влияет на существующие файлы;
    • policyName – политика erasure coding, используемая для файлов в этом каталоге. Параметр может быть пропущен, если установлена конфигурация dfs.namenode.ec.system.default.policy. Политика ЕС пути устанавливается со значением по умолчанию в конфигурации;
    • -replicate – применение схемы по умолчанию REPLICATION для каталога, принятие каталогом репликации ;
    • -replicate и -policy <policyName> опциональные аргументы, и они не могут быть указаны одновременно.
  • [-getPolicy -path <path>] – получение подробной информации о политике erasure coding файла или каталога по указанному пути;
  • [-unsetPolicy -path <path>] – сброс политики erasure coding в каталоге, заданной вызовом setPolicy. Если директория наследует политику от родительского каталога верхнего уровня, то операция недопустима. Сброс политики для каталога, в котором нет явного набора политик, не возвращает ошибку;
  • [-listPolicies] – перечисляет все (включенные, отключенные и удаленные) политики erasure coding, зарегистрированные в HDFS. Только включенные политики подходят для использования с командой setPolicy;
  • [-addPolicies -policyFile <file>] – добавление списка пользовательских политик erasure coding. Пример политики приведен в файле etc/hadoop/user_ec_policies.xml.template. Максимальный размер ячейки определяется в свойстве dfs.namenode.ec.policies.max.cellsize со значением по умолчанию 4 МБ. В настоящее время HDFS позволяет пользователю добавлять в общей сложности 64 политики, а ID добавленной политики находится в диапазоне от 64 до 127. Если уже существует 64 политики, то добавление новой завершается ошибкой;
  • [-listCodecs] – получение списка поддерживаемых кодеков erasure coding и кодеров в системе. Кодер – это реализация кодека. Кодек может иметь разные реализации, поэтому существуют разные кодеры. Кодеры для кодека перечисляются в обратном порядке;
  • [-removePolicy -policy <policyName>] – удаление пользовательской политики erasure coding;
  • [-enablePolicy -policy <policyName>] – включение политики erasure coding;
  • [-disablePolicy -policy <policyName>] – отключение политики erasure coding.

Ограничения

Некоторые операции HDFS, такие как hflush, hsync, concat, setReplication, truncate и append, не поддерживаются файлами erasure coding из-за существенных технических проблем:

  • append() и truncate() для файла erasure coding вызывают исключение IOException;
  • concat() вызывает исключение IOException, если файлы смешаны с разными политиками erasure coding или с реплицированными файлами;
  • setReplication() не работает для файлов erasure coding;
  • hflush() и hsync() для DFSStripedOutputStream не используются. Таким образом, вызов hflush() или hsync() для файла erasure coding не может гарантировать сохранность данных.

Клиент может использовать StreamCapabilities API для запроса, поддерживает ли OutputStream операции hflush() и hsync(). Если клиенту требуется постоянство данных с помощью этих функций, текущим решением является создание таких файлов, как обычные файлы репликации 3x, в каталоге без erasure coding, или использование FSDataOutputStreamBuilder#replicate() API для создания файлов репликации 3x в каталоге erasure coding.