Hadoop
Cистема Hadoop является наиболее популярной реализацией с открытыми кодами среды MapReduce, разрабатываемой Yahoo! и Apache Software Foundation . В отличие от реализации Google исходной среды MR, где использовался язык C++, ядро системы Hadoop целиком написано на языке Java. В экспериментах, описываемых в этой статье, использовалась система Hadoop версии 0.19.0, исполняемая в среде Java 1.6.0. Была установлена система с конфигурацией по умолчанию, за исключением следующих изменений, которые было решено внести для улучшения производительности, не отклоняясь от основных принципов ядра MR:
- данные хранились в блоках размеров в 256 мегабайт вместо 64 мегабайт, используемых по умолчанию;
- каждый исполнитель задач JVM запускался с максимальным размером кучи в 512 мегабайт, и DataNode/JobTracker виртуальной машины запускался с максимальным размеров кучи в 1024 мегабайт (при общем объеме основной памяти в 3,5 гигабайт на узел);
- в Hadoop была разрешена опция «rack awareness» (возможность учета физического размещения узлов при планировании задач) для обеспечения локальности данных в кластере;
- в Hadoop было разрешено повторно использовать исполнитель задач JVM, а не запускать новый процесс для каждой задачи Map/Reduce.
Кроме того, система была сконфигурирована таким образом, чтобы на каждом узле запускались два экземпляра Map и один экземпляр Reduce.
В среде Hadoop также обеспечивается некоторая реализация распределенной файловой системы Google . При каждом прогоне тестов все входные и выходные файлы сохранялись в распределенной файловой системе Hadoop (Hadoop distributed file system, HDFS). Использовались параметры HDFS по умолчанию с тремя репликами каждого блока и без сжатия. Тестировались и другие конфигурации (без репликации, со сжатием на уровнях блока и записи), но было обнаружено, что в этих условиях тесты выполняются с той же или худшей скоростью (см. п. 5.1.3). После завершения прогона тестов для заданного уровня масштабирования узлов каталоги данных на каждом узле удалялись, и HDFS форматировалась заново, чтобы следующий набор вводных данных реплицировался по уздам равномерно.
Для координации активности в узлах кластера в Hadoop используются центральный трекер заданий и «главный» (master) демон HDFS. Чтобы эти демоны гарантированно не влияли на производительность узлов-обработчиков, оба эти дополнительных компонента среды выполнялись в отдельном узле кластера.
Имеются два способа загрузки данных в распределенную файловую систему Hadoop: (1) использование файловой утилиты с интерфейсом командной строки для выгрузки в HDFS файлов, хранимых в локальной файловой системе, и (2) создание собственной программы загрузки данных, которая записывает данные с использованием внутреннего API ввода-вывода Hadoop. В данном случае не требовалось изменять вводные данные для тестовых MR-программ, и поэтому во всех узлы файлы загружались в HDFS параллельно в виде плоского текста с использованием утилиты командной строки. Хранение данных в такой манере позволяет MR-программам производить доступ к данным с использованием формата данных Hadoop TextInputFormat, в котором в каждом файле ключами являются номера строк, а соответствующие им значения – это содержимое строк. Было установлено, что этот подход приводит к более высокой эффективности как загрузки данных, так и выполнения задач, чем использование сериализованных форматов или средств сжатия данных Hadoop.
В отличие от набора данных задачи Grep, который загружался в HDFS в неизменном виде, наборы данных UserVisits и Rankings требовалось модифицировать, чтобы первый и второй столбцы разделялись символом табуляции, а все остальные поля каждой строки – некоторым уникальным разделителем полей. Поскольку в модели MR нет схем, для обеспечения доступа к различным атрибутам во время выполнения функции Map и Reduce в каждой задаче должны вручную разбивать значение в массив строк, руководствуясь символом-разделителем.
Был написал специальный загрузчик данных, выполняемый параллельно в каждом узле; этот загрузчик считывал строки наборов данных, подготавливал данные, как требовалось, и затем записывал полученный кортеж в плоский текстовый файл в HDFS. Загрузка данных таким способом происходила примерно в три раза медленнее, чем если бы использовалась утилита командной строки, но зато не потребовалось писать для Hadoop специальные обработчики ввода; имеется возможность использовать в MR-программах интерфейс KeyValueTextInputFormat, позволяющий автоматически расщеплять строки текстовых файлов на пары «ключ/значение» по символу табуляции. Было обнаружено, что использование других вариантов форматирования данных, таких как SequenceFileInputFormat или специальные Writable tuples, замедляет и загрузку, и исполнение программы.
