Apache Hadoop

Материал из Национальной библиотеки им. Н. Э. Баумана
Последнее изменение этой страницы: 01:38, 2 октября 2018.
Apache Hadoop
Apache Hadoop
Разработчики: Contributors
Постоянный выпуск: 2.7.3 / 25 August 2016 года; 2 years ago (2016-08-25)
Состояние разработки: Active
Написана на: Java
Операционная система: Cross-platform
Тип ПО: Data warehouse
Лицензия: Apache License 2.0
Веб-сайт hadoop.apache.org

Apache Hadoop — проект фонда Apache Software Foundation, свободно распространяемый набор утилит, библиотек и фреймворк для разработки и выполнения распределённых программ, работающих на кластерах из сотен и тысяч узлов. Используется для реализации поисковых и контекстных механизмов многих высоконагруженных веб-сайтов, в том числе, для Yahoo! и Facebook. Разработан на Java в рамках вычислительной парадигмы MapReduce, согласно которой приложение разделяется на большое количество одинаковых элементарных заданий, выполнимых на узлах кластера и естественным образом сводимых в конечный результат.

По состоянию на 2014 год проект состоит из четырёх модулей — Hadoop Common (связующее программное обеспечение — набор инфраструктурных программных библиотек и утилит, используемых для других модулей и родственных проектов), HDFS (распределённая файловая система), YARN (система для планирования заданий и управления кластером) и Hadoop MapReduce (платформа программирования и выполнения распределённых MapReduce-вычислений), ранее в Hadoop входил целый ряд других проектов, ставших самостоятельными в рамках системы проектов Apache Software Foundation.

Считается одной из основополагающих технологий «больших данных». Вокруг Hadoop образовалась целая экосистема из связанных проектов и технологий, многие из которых развивались изначально в рамках проекта, а впоследствии стали самостоятельными. Со второй половины 2000-х годов идёт процесс активной коммерциализации технологии, несколько компаний строят бизнес целиком на создании коммерческих дистрибутивов Hadoop и услуг по технической поддержке экосистемы, а практически все крупные поставщики информационных технологий для организаций в том или ином виде включают Hadoop в продуктовые стратегии и линейки решений.

История

Разработка была инициирована в начале 2005 года Дугом Каттингом с целью построения программной инфраструктуры распределённых вычислений для проекта Nutch — свободной программной поисковой машины на Java, её идейной основой стала публикация сотрудников Google Джеффри Дина и Санжая Гемавата о вычислительной концепции MapReduce. Новый проект был назван в честь игрушечного слонёнка ребёнка основателя проекта.

В течение 2005—2006 годов Hadoop развивался усилиями двух разработчиков — Каттинга и Майка Кафареллы (Mike Cafarella) в режиме частичной занятости, сначала в рамках проекта Nutch, затем — проекта Lucene. В январе 2008 года Hadoop становится проектом верхнего уровня системы проектов Apache Software Foundation. В апреле 2008 года Hadoop побил мировой рекорд производительности в стандартизованном бенчмарке сортировки данных — 1 Тбайт был обработан за 209 сек. на кластере из 910 узлов. В апреле 2010 года корпорация Google предоставила Apache Software Foundation права на использование технологии MapReduce, через три месяца после её защиты в патентном бюро США, тем самым избавив организацию от возможных патентных претензий.

Начиная с 2010 года Hadoop неоднократно характеризуется как ключевая технология «больших данных», прогнозируется его широкое распространение для массово-параллельной обработки данных, и, наряду с Cloudera, появилась серия технологических стартапов, целиком ориентированных на коммерциализацию Hadoop. В течение 2010 года несколько подпроектов Hadoop — Avro, HBase, Hive, Pig, Zookeeper — последовательно стали проектами верхнего уровня фонда Apache, что послужило началом формирования экосистемы вокруг Hadoop. В марте 2011 года Hadoop удостоен ежегодной инновационной награды медиагруппы Guardian, на церемонии вручения технология была названа «швейцарским армейским ножом XXI века».

Реализация в вышедшем осенью 2013 года Hadoop 2.0 модуля YARN оценена как значительный скачок, выводящий Hadoop за рамки парадигмы MapReduce и ставящая технологию на уровень универсального решения для организации распределённой обработки данных.

Возможности

Изначально Hadoop был, в первую очередь, инструментом для хранения данных и запуска MapReduce-задач, сейчас же Hadoop представляет собой большой стек технологий, так или иначе связанных с обработкой больших данных (не только при помощи MapReduce).

Основными (core) компонентами Hadoop являются:

  • Hadoop Distributed File System (HDFS) – распределённая файловая система, позволяющая хранить информацию практически неограниченного объёма.
  • Hadoop YARN – фреймворк для управления ресурсами кластера и менеджмента задач, в том числе включает фреймворк MapReduce.
  • Hadoop common

Также существует большое количество проектов непосредственно связанных с Hadoop, но не входящих в Hadoop core:

  • Hive – инструмент для SQL-like запросов над большими данными (превращает SQL-запросы в серию MapReduce–задач);
  • Pig – язык программирования для анализа данных на высоком уровне. Одна строчка кода на этом языке может превратиться в последовательность MapReduce-задач;
  • HBase – колоночная база данных, реализующая парадигму BigTable;
  • Cassandra – высокопроизводительная распределенная key-value база данных;
  • ZooKeeper – сервис для распределённого хранения конфигурации и синхронизации изменений этой конфигурации;
  • Mahout – библиотека и движок машинного обучения на больших данных.

Отдельно хотелось бы отметить проект Apache Spark, который представляет собой движок для распределённой обработки данных. Apache Spark обычно использует компоненты Hadoop, такие как HDFS и YARN для своей работы, при этом сам в последнее время стал популярнее, чем Hadoop:

Динамика популярности запросов "Apache Hadoop" и "Apache Spark" за период с 2004 года по конец 2016 года

Масштабируемость

Одной из основных целей Hadoop изначально было обеспечение горизонтальной масштабируемости кластера посредством добавления недорогих узлов, без использования к мощных серверов и дорогих сетей хранения данных. Функционирующие кластеры размером в тысячи узлов подтверждают осуществимость и экономическую эффективность таких систем, так, по состоянию на 2011 год известно о крупных кластерах Hadoop в Yahoo (более 4 тыс. узлов с суммарной ёмкостью хранения 15 Пбайт), Facebook (около 2 тыс. узлов на 21 Пбайт) и Ebay (700 узлов на 16 Пбайт)[1]. Тем не менее, считается, что горизонтальная масштабируемость в Hadoop-системах ограничена, для Hadoop до версии 2.0 максимально возможно оценивалась в 4 тыс. узлов при использовании 10 MapReduce-заданий на узел. Во многом этому ограничению способствовала концентрация в модуле MapReduce функций по контролю за жизненным циклом заданий, считается, что с выносом её в модуль YARN в Hadoop 2.0 и децентрализацией — распределением части функций по мониторингу на узлы обработки — горизонтальная масштабируемость повысилась.

Ещё одним ограничением Hadoop-систем является размер оперативной памяти на узле имён, хранящем всё пространство имён кластера для распределения обработки, притом общее количество файлов, которое способен обрабатывать узел имён — 100 млн. Для преодоления этого ограничения ведутся работы по распределению узла имён, единого в текущей архитектуре на весь кластер, на несколько независимых узлов. Другим вариантом преодоления этого ограничения является использование распределённых СУБД поверх HDFS, таких как HBase, роль файлов и каталогов в которых с точки зрения приложения играют записи в одной большой таблице базы данных.

По состоянию на 2011 год типичный кластер строился из однопроцессорных многоядерных x86-64-узлов под управлением Linux с 3-12 дисковыми устройствами хранения, связанных сетью с пропускной способностью 1 Гбит/с. Существуют тенденции как к снижению вычислительной мощности узлов и использованию процессоров с низким энергопотреблением (ARM, Intel Atom), так и применения высокопроизводительных вычислительных узлов одновременно с сетевыми решениями с высокой пропускной способностью (InfiniBand в Oracle Big Data Appliance[en], высокопроизводительная сеть хранения данных на Fibre Channel и Ethernet пропускной способностью 10 Гбит/с в шаблонных конфигурациях FlexPod для «больших данных»).

Масштабируемость Hadoop-систем в значительной степени зависит от характеристик обрабатываемых данных, прежде всего, их внутренней структуры и особенностей по извлечению из них необходимой информации, и сложности задачи по обработке, которые, в свою очередь, диктуют организацию циклов обработки, вычислительную интенсивность атомарных операций, и, в конечном счёте, уровень параллелизма и загруженность кластера. В руководстве Hadoop (первых версий, ранее 2.0) указывалось, что приемлемым уровнем параллелизма является использование 10-100 экземпляров базовых обработчиков на узел кластера, а для задач, не требующих значительных затрат процессорного времени — до 300; для свёрток считалось оптимальным использование их по количеству узлов, умноженному на коэффициент из диапазона от 0,95 до 1,75. С большим значением коэффициента наиболее быстрые узлы, закончив первый раунд сведения, раньше получат вторую порцию промежуточных пар для обработки, таким образом, увеличение коэффициента избыточно загружает кластер, но при этом обеспечивает более эффективную балансировку нагрузки. В YARN вместо этого используются конфигурационные константы, определяющие значения доступной оперативной памяти и виртуальных процессорных ядер, доступных для планировщика ресурсов, на основании которых и определяется уровень параллелизма.

Hadoop MapReduce

Hadoop MapReduce — программный каркас для программирования распределённых вычислений в рамках парадигмы MapReduce. Разработчику приложения для Hadoop MapReduce необходимо реализовать базовый обработчик, который на каждом вычислительном узле кластера обеспечит преобразование исходных пар «ключ — значение» в промежуточный набор пар «ключ — значение» (класс, реализующий интерфейс Mapper, назван по функции высшего порядка Map), и обработчик, сводящий промежуточный набор пар в окончательный, сокращённый набор (свёртку, класс, реализующий интерфейс Reducer). Каркас передаёт на вход свёртки отсортированные выводы от базовых обработчиков, сведе́ние состоит из трёх фаз:

  • shuffle (тасовка, выделение нужной секции вывода)
  • sort (сортировка, группировка по ключам выводов от распределителей — досортировка, требующаяся в случае, когда разные атомарные обработчики возвращают наборы с одинаковыми ключами, при этом, правила сортировки на этой фазе могут быть заданы программно и использовать какие-либо особенности внутренней структуры ключей)
  • reduce (свёртка списка) — получения результирующего набора. Для некоторых видов обработки свёртка не требуется, и каркас возвращает в этом случае набор отсортированных пар, полученных базовыми обработчиками.

Hadoop MapReduce позволяет создавать задания как с базовыми обработчиками, так и со свёртками, написанными без использования Java: утилиты Hadoop streaming позволяют использовать в качестве базовых обработчиков и свёрток любой исполняемый файл, работающий со стандартным вводом-выводом операционной системы (например, утилиты командной оболочки UNIX), есть также SWIG-совместимый прикладной интерфейс программирования Hadoop pipes на C++. Также, в состав дистрибутивов Hadoop входят реализации различных конкретных базовых обработчиков и свёрток, наиболее типично используемых в распределённой обработке.

В первых версиях Hadoop MapReduce включал планировщик заданий (JobTracker), начиная с версии 2.0 эта функция перенесена в YARN, и начиная с этой версии модуль Hadoop MapReduce реализован поверх YARN. Программные интерфейсы по большей части сохранены, однако полной обратной совместимости нет (то есть для запуска программ, написанных для предыдущих версий API, для работы в YARN в общем случае требуется их модификация или рефакторинг, и лишь при некоторых ограничениях возможны варианты обратной двоичной совместимости).

Использование

Теперь покажем как запустить MapReduce-задачу на Hadoop. В качестве задачи воспользуемся классическим примером WordCount, который был разобран в предыдущей статье цикла. Напомню формулировку задачи: имеется набор документов. Необходимо для каждого слова, встречающегося в наборе документов, посчитать, сколько раз встречается слово в наборе.

Решение: Map разбивает документ на слова и возвращает множество пар (word, 1). Reduce суммирует вхождения каждого слова:

function map *(source) {
    for (const word in source.split(' ')) {
        yield [word, 1];
    }
}
function reduce *(word, values) {
    yield [word, values.length];
}

Теперь задача запрограммировать это решение в виде кода, который можно будет исполнить на Hadoop и запустить. Самый простой способ запустить MapReduce-программу на Hadoop — воспользоваться streaming-интерфейсом Hadoop. Streaming-интерфейс предполагает, что map и reduce реализованы в виде программ, которые принимают данные с stdin и выдают результат на stdout.

Программа, которая исполняет функцию map называется mapper. Программа, которая выполняет reduce, называется, соответственно, reducer. Streaming интерфейс предполагает по умолчанию, что одна входящая строка в mapper или reducer соответствует одной входящей записи для map. Вывод mapper’a попадает на вход reducer’у в виде пар (ключ, значение), при этом все пары соответствующие одному ключу:

  • Гарантированно будут обработаны одним запуском reducer’a;
  • Будут поданы на вход подряд (то есть если один reducer обрабатывает несколько разных ключей – вход будет сгруппирован по ключу).

Итак, реализуем mapper и reducer на JavaScript:

// mapper.js
process.stdin.setEncoding('utf8');
let input = '';
process.stdin.on('readable', function () {
    const chunk = process.stdin.read();
    if (chunk !== null) {
        input += chunk;
    }
});

process.stdin.on('end', function () {
    input
        .split(' ')
        .forEach(word => process.stdout.write(`${word}\t1\n`));
});
// reducer.js
process.stdin.setEncoding('utf8');
let input = '';
process.stdin.on('readable', function () {
    const chunk = process.stdin.read();
    if (chunk !== null) {
        input += chunk;
    }
});

process.stdin.on('end', function () {
    const result = input
        .split('\n')
        .reduce((result, line) => {
            const [word, value] = line.split('\t');
            result[word] = (result[word] || 0) + (+value);
            return result;
        }, {});
    Object.keys(result)
        .forEach(word => process.stdout.write(`${word}\t${result[word]}\n`));
});

Данные, которые будет обрабатывать Hadoop должны храниться на HDFS. Загрузим наши статьи и положим на HDFS. Для этого нужно воспользоваться командой hadoop fs:

$ hadoop fs -put input_text.txt

Теперь запустим streaming-задачу:

$ yarn jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar\ 
 -input input_text.txt\ 
 -output output\ 
 -file mapper.js\ 
 -file reducer.js\ 
 -mapper "node mapper.js"\ 
 -reducer "node reducer.js"

Утилита yarn служит для запуска и управления различными приложениями (в том числе map-reduce based) на кластере. hadoop-streaming.jar — это как раз один из примеров такого yarn-приложения.

Дальше идут параметры запуска:

  • input – папка с исходными данными на hdfs;
  • output – папка на hdfs, куда нужно положить результат;
  • file – файлы, которые нужны в процессе работы map-reduce задачи;
  • mapper – консольная команда, которая будет использоваться для map-стадии;
  • reduce – консольная команда которая будет использоваться для reduce-стадии.

После запуска в консоли можно будет увидеть прогресс выполнения задачи и URL для просмотра более детальной информации о задаче. В интерфейсе доступном по этому URL можно узнать более детальный статус выполнения задачи, посмотреть логи каждого маппера и редьюсера (что очень полезно в случае упавших задач).

Прогресс выполнения задачи map-reduce

Результат работы после успешного выполнения складывается на HDFS в папку, которую мы указали в поле output. Просмотреть её содержание можно при помощи команды hadoop fs -ls lenta_wordcount.

Установка

  • Обновить данные о репозиториях
    $ sudo apt-get update
    $ sudo apt-get upgrade
  • Скачаем официальный дистрибутив с зеркала и распакуем его
    $ wget http://apache-mirror.rbc.ru/pub/apache/hadoop/common/hadoop-2.7.3/hadoop-2.7.3.tar.gz
    $ tar -zxf hadoop-2.7.3.tar.gz
  • Проверим корректность установки
    $ cd hadoop-2.7.3/
    $ ./bin/hadoop

Источники

  1. Shvachko, Konstantin Apache Hadoop. The Scalability Update, 2011, Yahoo reportedly ran numerous clusters having 4000+ nodes with four 1 TB drives per node, 15 PB of total storage capacity, 70 million files, and 80 million blocks using 50 GB NameNode heap. Facebook’s 2000-node warehouse cluster is provisioned for 21 PB of total storage capacity. Extrapolating the announced growth rate, its namespace should have close to 200 million objects (files + blocks) by now, but an immense 108 GB heap should allow room for close to 400 million objects. eBay runs a 700-node cluster. Each node has 24 TB of local disk storage, 72 GB of RAM, and a 12-core CPU. Total cluster size is 16 PB.

Ссылки