Версия сайта для слабовидящих
Мы уже писали, что технологии Big Data ориентированы на работу с большими данными, а не множеством маленьких. Сегодня рассмотрим подробнее, почему Apache Hadoop, Spark и основанные на HDFS NoSQL-СУБД Hive и HBase плохо работают с большим количеством маленьких файлов, а также как это исправить.
Почему HDFS плохо работает со множеством маленьких файлов
Прежде всего, отметим, файлы какого размера считаются маленькими и относительно чего выставляется такая оценка. Эталоном сравнения выступает блок распределенной файловой системы Apache Hadoop (HDFS) размером 64 или 128 МБ. Однако, блоки HDFS являются не физической, а логической единицей распределения хранилища. В HDFS каждый файл хранится в таком блоке, а метаданные блока хранятся в памяти узла имен в виде объекта. Файлы и блоки являются объектами имен в HDFS и занимают пространство имен, т.е. место в NameNode. Это означает, что емкость пространства имен системы ограничена физической памятью узла имен. Примерный размер для хранения каждого такого объекта равен 150 байт. Поэтому, например, 20 миллионов файлов займут 3 ГБ памяти. А масштабирование энергозависимой памяти (ОЗУ) сегодня является проблемой.
Будучи базовой технологией Big Data, HDFS предназначена для хранения и обработки больших массивов данных порядка ТБ. Например, кластер Apache Hadoop может иметь 14 ПБ дискового пространства и хранить 60 миллионов файлов. А вот небольшие файлы HDFS хранит совсем неэффективно: много мелких файлов будут занимать много места на NameNode, хотя сами небольшие файлы не занимают больше места на диске, чем требуется для хранения их содержимого. Например, файл размером 1 МБ хранится с размером блока 128 МБ, но использует ровно 1 МБ дискового пространства, а не 128 МБ. На практике небольшие файлы могут занимать до 95% пространства имен, фактически располагаюсь на 30% дискового пространства кластера.
Таким образом, все файлы, которые меньше размера блока HDFS считаются маленькими, и работа с такими объемами становится неэффективной из-за неэффективной утилизации пространства блока. Это характерно для всей экосистемы Apache Hadoop, работающей с HDFS, включая вычислительный движок Spark, а также NoSQL-СУБД HBase и Hive. Например, при записи в таблицу Hive с динамическим разделением каждый раздел обрабатывается исполнителями параллельно. При этом, когда исполнитель встречает новый ключ раздела в текущем, он открывает новый файл. По умолчанию Spark использует для данных разделитель Hash или Round Robin, применение которых к произвольному датафрейму распределяет строки относительно равномерно, но случайным образом по разделам. Поэтому без дополнительных действий при произвольной записи данных создается примерно 1 файл на раздел для каждого уникального ключа раздела.
В результате этого множество маленьких файлов, неэффективно использующих HDFS, приводит к следующим последствиям:
- замедление скорости обработки, т.е. выполнения заданий Spark, MapReduce и Hive;
- замедление чтения — маленькие файлы требуют многократных попыток извлечения данных;
- рост рабочей нагрузки – небольшие файлы размером несколько КБ создаются тысячами при одновременном выполнении заданий;
- падение производительности – Mapper обычно обрабатывает 1 блок ввода за раз, а из-за большого количества мелких файлов для выполнения работы требуется больше сопоставлений;
- увеличение дисковых операций ввода-вывода, поскольку для записи и поиска множества маленьких файлов много времени уходит на сканирование дискового пространства;
- устаревание данных — низкая скорость чтения и обработки, а также бесполезная трата памяти приводят к созданию устаревших данных, что затрудняет весь процесс отчетности и аналитики.
Тем не менее, на практике файлы размером меньше блока HDFS встречаются довольно часто. Их создают потоковые данные, большое количество заданий MapReduce, партиционированные таблицы с несколькими разделами, чрезмерное распараллеливание заданий, а также специализированные форматы файлов и сжатие. Как избежать отмеченных проблем при работе Hadoop с такими данными или смягчить их последствия, рассмотрим далее.
Как бороться с проблемой маленьких файлов в Apache Hadoop
Одним из способов избежать проблемы маленьких файлов в HDFS является использование архивов HAR, которые располагаются поверх HDFS. Команда HAR создаст файл HAR-архива, который затем запустит задание MapReduce, чтобы упаковать много небольших файлов в один большой. При этом HAR обеспечивает прозрачный доступ к исходным файлам и увеличивает масштабируемость системы за счет сокращения использования пространства имен и уменьшения рабочих нагрузок в NameNode. Дополнительно HAR оптимизирует утилизацию памяти и распределяет управление пространством имен по нескольким NameNode, а также распараллеливает доступ к исходным файлам с помощью заданий MapReduce. А поскольку HAR-архивы отображаются как файловая система, MapReduce может использовать все логические входные файлы в качестве входных данных.
Другим способом борьбы с множеством маленьких файлов в HDFS является использование формата последовательности (Sequence File), где имя файла представляет собой ключ, а содержимое – его значение. Можно упаковать много небольших файлов в один файл последовательности, который является разделяемым. Поэтому одно задание MapReduce может работать с каждым фрагментом независимо и параллельно.
В Apache HBase большие файлы создаются в зависимости от схемы доступа к небольшим файлам с помощью мажорного или минорного сжатия (Major/Minor Compaction), о чем мы писали здесь. А вот в Apache Hive это немного сложнее. В этой NoSQL-СУБД количество файлов в разделе растет по мере частых обновлений таблицы и при неравномерном распределении, когда количество используемых преобразователей находится на более нагруженной стороне. Например, если потоковые данные загружаются в таблицу Hive с коротким промежутком времени. Чтобы снизить негативное влияние на HDFS, рекомендуется настроить конфигурации Hive следующим образом:
set hive.exec.compress.output=true; set hive.exec.parallel = true; set parquet.compression=snappy; set hive.merge.mapfiles=true; set hive.merge.mapredfiles=true; set hive.merge.smallfiles.avgsize = 134217728; — 128M set hive.merge.size.per.task = 268435456; — 256M set hive.optimize.sort.dynamic.partition = true; set parquet.blocksize= 268435456; — 256M set dfs.block.size=268435456;
Можно установить размер блока колоночного формата Parquet для вывода файлов желаемой длины при вставке данных в Hive или Impala, попутно установив размер блока HDFS. Если в датафрейме нет записей, а выполняется вставка в таблицу Hive, создается файл с метаданными и нулевыми записями, что может привести к созданию множества файлов метаданных. Поэтому перед операцией записи необходимо проверить датафрейм на наличие значений (Not NULL).
В отличие от обычной таблицы Hive, ACID-таблица выполняет сжатие автоматически, если включены свойства compactor.mapreduce.map.memory.mb, compactorthreshold.hive.compactor.delta.num.threshold и compactorthreshold.hive.compactor.delta.pct.threshold.
Вычислительной движок Apache Spark тоже не очень хорошо работает с множеством маленьких файлов. Но при загрузке данных в системы хранения объектов, такие как HDFS, AWS S3 и пр., может случиться ситуация с генерацией большого количества небольших файлов, т.к. Spark — это система параллельной обработки, а загрузка данных выполняется с помощью нескольких задач, где каждая задача может загружаться в несколько разделов.
Например, если для конфигурации spark.sql.shuffle.partitions установлено значение 200, а выражение Partition By используется для загрузки, в 50 целевых разделов, будет 200 задач загрузки, каждая из которых потенциально может загружаться в 50 разделов. В результате каждая загрузка создаст около 200*50=10000 файлов.
Справиться с этим помогут методы coalesce() и repartition(), про которые мы рассказывали здесь. Оба этих метода можно использовать для контроля количества генерируемых файлов, но repartition() является более дорогостоящей операцией, поскольку требует перетасовки всех данных. Это перераспределение можно использовать для создания файлов желаемого размера путем уменьшения или увеличения количества разделов в датафрейме, а coalesce() уменьшает количество разделов. Рассчитать оптимальное количество файлов можно разделив общий размер файла на настроенный размер блока HDFS. Также можно использовать настроить конфигурацию shuffle.partition, по умолчанию равную 200. Изменение этого значения в соответствии с решаемой задачей также помогает уменьшить количество небольших файлов.
Сжатие или уплотнение можно использовать для решения проблем с небольшими файлами путем объединения небольших файлов. Для этого дата-инженер может самостоятельно написать код для утилиты сжатия на уровне папки таблицы и сжимать файлы, используя Spark для слияния файлов в папку. Для этого следует рассчитать общий размер файлов для объединения и оптимальное количество разделов, разделив общий размер файлов на размер блока HDFS. Далее следует передать нужный раздел для объединения, а затем удалить файлы метаданных. При этом рекомендуется рассматривать для слияния только те файлы, размер которых меньше размера блока HDFS. Это предотвратит повторное слияние уже объединенных файлов или файлов, размер которых превышает размер блока. А вот использовать слияние в Parquet-файлах не рекомендуется, поскольку это чревато снижением производительности. Читайте в нашей новой статье про альтернативную HDFS платформу организации озера данных в облаке — MinIO и о том, как Apache Spark работает с ней.
Больше подробностей про администрирование и эксплуатацию Apache Hadoop для хранения и аналитики больших данных вы узнаете на специализированных курсах в нашем лицензированном учебном центре обучения и повышения квалификации для разработчиков, менеджеров, архитекторов, инженеров, администраторов, Data Scientist’ов и аналитиков Big Data в Москве:
Источники
