Что такое mapreduce и как он работает

Big data от А до Я. Часть 3: Приемы и стратегии разработки MapReduce-приложений

Привет, Хабр! В предыдущих статьях мы описали парадигму MapReduce, а также показали как на практике реализовать и выполнить MapReduce-приложение на стеке Hadoop. Пришла пора описать различные приёмы, которые позволяют эффективно использовать MapReduce для решения практических задач, а также показать некоторые особенности Hadoop, которые позволяют упростить разработку или существенно ускорить выполнение MapReduce-задачи на кластере.

Map only job

Как мы помним, MapReduce состоит из стадий Map, Shuffle и Reduce. Как правило, в практических задачах самой тяжёлой оказывается стадия Shuffle, так как на этой стадии происходит сортировка данных. На самом деле существует ряд задач, в которых можно обойтись только стадией Map. Вот примеры таких задач:

Пример конфигурации map-only задачи на hadoop:

Более развёрнутый пример по ссылке.Не указываем редьюсер и указываем нулевое количество редьюсеров Пример:

Map Only jobs на самом деле могут быть очень полезными. Например, в платформе Facetz.DCA для выявления характеристик пользователей по их поведению используется именно один большой map-only, каждый маппер которого принимает на вход пользователя и на выход отдаёт его характеристики.

Combine

Как я уже писал, обычно самая тяжёлая стадия при выполнении Map-Reduce задачи – это стадия shuffle. Происходит это потому, что промежуточные результаты (выход mapper’a) записываются на диск, сортируются и передаются по сети. Однако существуют задачи, в которых такое поведение кажется не очень разумным. Например, в той же задаче подсчёта слов в документах можно предварительно предагрегировать результаты выходов нескольких mapper’ов на одном узле map-reduce задачи, и передавать на reducer уже просуммированные значения по каждой машине.

В hadoop для этого можно определить комбинирующую функцию, которая будет обрабатывать выход части mapper-ов. Комбинирующая функция очень похожа на reduce – она принимает на вход выход части mapper’ов и выдаёт агрегированный результат для этих mapper’ов, поэтому очень часто reducer используют и как combiner. Важное отличие от reduce – на комбинирующую функцию попадают не все значения, соответствующие одному ключу.

Более того, hadoop не гарантирует того, что комбинирующая функция вообще будет выполнена для выхода mapper’a. Поэтому комбинирующая функция не всегда применима, например, в случае поиска медианного значения по ключу. Тем не менее, в тех задачах, где комбинирующая функция применима, её использование позволяет добиться существенного прироста к скорости выполнения MapReduce-задачи.

Использование Combiner’a на hadoop:

Цепочки MapReduce-задач

Бывают ситуации, когда для решения задачи одним MapReduce не обойтись. Например, рассмотрим немного видоизмененную задачу WordCount: имеется набор текстовых документов, необходимо посчитать, сколько слов встретилось от 1 до 1000 раз в наборе, сколько слов от 1001 до 2000, сколько от 2001 до 3000 и так далее.

Для решения нам потребуется 2 MapReduce job’а:

Для того, чтобы выполнить последовательность MapReduce-задач на hadoop, достаточно просто в качестве входных данных для второй задачи указать папку, которая была указана в качестве output для первой и запустить их по очереди.

На практике цепочки MapReduce-задач могут представлять собой достаточно сложные последовательности, в которых MapReduce-задачи могут быть подключены как последовательно, так и параллельно друг другу. Для упрощения управления такими планами выполнения задач существуют отдельные инструменты типа oozie и luigi, которым будет посвящена отдельная статья данного цикла.

Пример цепочки MapReduce-задач.

Distributed cache

Важным механизмом в Hadoop является Distributed Cache. Distributed Cache позволяет добавлять файлы (например, текстовые файлы, архивы, jar-файлы) к окружению, в котором выполняется MapReduce-задача.

Можно добавлять файлы, хранящиеся на HDFS, локальные файлы (локальные для той машины, с которой выполняется запуск задачи). Я уже неявно показывал, как использовать Distributed Cache вместе с hadoop streaming: добавляя через опцию -file файлы mapper.py и reducer.py. На самом деле можно добавлять не только mapper.py и reducer.py, а вообще произвольные файлы, и потом пользоваться ими как будто они находятся в локальной папке.

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

Reduce Join

Те, кто привык работать с реляционными базами, часто пользуются очень удобной операцией Join, позволяющей совместно обработать содержание некоторых таблиц, объединив их по некоторому ключу. При работе с большими данными такая задача тоже иногда возникает. Рассмотрим следующий пример:

Имеются логи двух web-серверов, каждый лог имеет следующий вид:
Необходимо посчитать для каждого IP-адреса на какой из 2-х серверов он чаще заходил. Результат должен быть представлен в виде:
К сожалению, в отличие от реляционных баз данных, в общем случае объединение двух логов по ключу (в данном случае – по IP-адресу) представляет собой достаточно тяжёлую операцию и решается при помощи 3-х MapReduce и паттерна Reduce Join:

Общая схема ReduceJoin

ReduceJoin работает следующим образом:

Где key – это ключ, по которому нужно объединять таблицы, Type – тип таблицы (first или second в нашем случае), а Value – это любые дополнительные данные, привязанные к ключу.

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

MapJoin

Паттерн ReduceJoin описывает общий случай объединения двух логов по ключу. Однако есть частный случай, при котором задачу можно существенно упростить и ускорить. Это случай, при котором один из логов имеет размер существенно меньшего размера, чем другой. Рассмотрим следующую задачу:

Имеются 2 лога. Первый лог содержит лог web-cервера (такой же как в предыдущей задаче), второй файл (размером в 100кб) содержит соответствие URL-> Тематика. Пример 2-го файла:

Для каждого IP-адреса необходимо рассчитать страницы какой категории с данного IP-адреса загружались чаще всего.

Далее задача решается следующим образом:

Reduce получает на вход ip и список всех тематик, просто вычисляет, какая из тематик встретилась чаще всего. Таким образом задача решена при помощи 1-го MapReduce, а собственно Join вообще происходит внутри map (поэтому если бы не нужна была дополнительная агрегация по ключу – можно было бы обойтись MapOnly job-ом):

Схема работы MapJoin

Резюме

В статье мы рассмотрели несколько паттернов и приемов решения задач при помощи MapReduce, показали, как объединять MapReduce-задачи в цепочки и join-ить логи по ключу.

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

Источник

MapReduce или подсчеты за пределами возможностей памяти и процессора (попробую без зауми)

Давно хотел рассказать про MapReduce, а то как ни взгляшешь на подобное — такая заумь, что просто ужас берет, а на самом деле очень простой и полезный подход для многих целей. И реализовать самому — не так уж и сложно.

Сразу скажу — топик — для тех, кто не разобрался что такое MapReduce. Для тех, кто разобрался — полезного тут ничего не будет.

Начнем с того как собственно родилась лично у меня идея MapReduce (хотя я и не знал, что он так называется, и, разумеется, пришла она мне куда позже чем Гугловсцам).

Сначала опишу как она рождалась (подход был неправильный), а потом как надо правильно делать.

Как посчитать все слова в Википедии (неправильный подход)

А родилась она, как и, наверное, везде — для подсчета частоты слов, когда обычной памяти не хватает (подсчет частоты всех слов в Википедии). Вместо слова «частота» тут скорее должно быть «количество вхождений», но для простоты оставлю «частота».

В самом простом случае мы можем завести хеш (dict, map, hash, ассоциативный массив, array() в PHP) и считать в нем слова.

Но что делать когда память под хеш кончится, а мы посчитали только одну сотую всех слов?

Я решил эту проблему тем, что считал часть слов, пока не кончится память, делал сохранение хеша на диск. То есть прямо построчно в файл:

aardvark | 5
aachen | 2

Возникла проблема — а как объединять-то эти файлы? Ведь каждый из них занимает всю оперативку.

Сначала была идея брать только самые популярные 1 000 000 слов из каждого файла и объединить их — это влезет в оперативку и посчитает хотя бы верхнюю часть списка (самые популярные слова). Это, конечно, сработало, но получилось что миллионы нижних слов терялись, а их было куда больше.

Пришла идея сортировать файлы.

Берем потом 20 сортированных файлов, читаем из каждого них первые 1000 строк, они будут примерно про одни и те же слова (отсортированные же файлы). Суммируем и формируем новый хеш, в нем будут только слова, начинающиеся на «aaa. » и подобные, сохраняем в новые файлы. Читаем следующие 1000 строк, все то же. Там примерно во всех файлах будут слова «aab. »

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

Муторно, долго, но лучше варианта в голову не пришло.

Слабое место неправильного подхода

В этом подходе было одно слабое место — а именно — объединение первоначальных 20 файлов. Как его сделать лучше?

Проблема возникает из того, что некоторых слов не будет в каких-то файлах или они будут в разных блоках по 1000 строк. То есть если бы я мог из всех 20 файлов брать не первые 1000 строк, а только по одной строке, но с одним и тем же словом — я бы смог все 20 файлов объединять за один проход.

Как это сделать? Вообще это ↑ последний шаг алгоритма MergeSort — объединение сортированных списков. Если знаете — пропускайте.

Берем по первой строке из всех 20 файлов, ищем минимальный первый элемент (слово) — он будет самый минимальный вообще во всем, поскольку у нас файлы отсортированы. Допустим это будет слово «aardvark» Берем из всех 20 строк, что мы прочитали только те, которые относятся к этому слову «aardvark». А оттуда, откуда мы его изымаем — только в тех файлах читаем вторую строку. Опять ищем минимальную среди этих 20. По аналогии — дальше, пока конца всех файлов не достигнем.

MapReduce в простейшем виде

Собственно, вот я почти и изобрел для себя то, что Google изобрело до меня десятилетие назад и назвало MapReduce.

Изобретение велосипедов продолжается и по сей день.

Шаг первый, берем строку, разбиваем на слова и выдаем вот такие массивы (ну или вернее: «tuples» — «кортежи»):

[ ‘foo’, 1 ]
[ ‘bar’, 1 ]
[ ‘baz’, 1 ]
[ ‘bar’, 1 ]

(Дальше буду скобочки и кавычки опускать там, где и так будет понятно)
Берем их, сортируем:

bar, 1
bar, 1
baz, 1
foo, 1

Замечаем, что bar идет два раза подряд, так что объединяем в такой вид:

Дальше просто складываем вторые элементы массивов. Получаем:

Именно то, что и хотели.

Главный вопрос — нафига козе баян… или что мы тут вообще делали и зачем?

Назад в прошлое

Если представить что у нас компьютер в который влезает только 2 строки и он может выполнять только одну операцию со строкой в минуту. (Отставить хихикать! После того как хотя бы раз посчитаете все слова в Википедии — будете иметь право смеяться над поставленными ограничениями памяти, все равно не влезет, хоть у Вас сколько гигов будет, а если влезет — считайте во всем Интернете 🙂 ).

Мы можем (из «foo bar baz bar» ) сделать два файла таким образом:

file1.txt
[ ‘bar’, 1 ]
[ ‘foo’, 1 ]

У нас в памяти по две строки — все в порядке, уложились в ограничения по памяти.

Теперь используя шаг из MergeSort, мы можем объединять построчно эти файлы:

При этом в памяти каждый раз у нас только две строки хранятся из 2 файлов — больше и не надо.

Собственно, то, что мы сделали — это уже MapReduce.

Тот шаг, который из слов выдает массивы с единичками ( слово, 1 ) — этот шаг называется «Map».
Тот шаг, который суммирует (1,1) — это шаг «Reduce».

Остальные шаги сделает сам алгоритм (сортировку и объединение через MergeSort).

Map, Reduce? Что это?

Сами эти шаги не обязательно заключаются в том, чтобы выдавать единички в случае «Map» или складывать в случае «Reduce». Это просто функции, которые могут что-то принять и что-то выдать. В зависимости от цели.

Проще говоря в Python:

Итак, мы обошли ограничение по памяти, но как обойти ограничение по скорости?

Представим что у нас два таких компьютера. Каждому из них мы даем исходную строку и говорим первому (точнее MapReduce говорит): считай только слова на нечетных местах, а второму — считай слова только на четных местах.

Первый выдает:
«foo bar baz bar»:
foo, 1
baz, 1

Второй выдает:
«foo bar baz bar»:
bar, 1
bar, 1

Мы (точнее MapReduce) забираем результаты у обоих, сортируем, потом прогоняем через MergeSort, как и выше:

Ровно тот же результат, как и когда считал один компьютер!

Теперь это мы (MapReduce) раздаем опять двум компьютерам: первому даем только нечетные строки, второму — четные и просим каждый компьютер делать шаг Reduce (сложить вторые цифры).

Собственно, понятно, что эти строки друг от друга не зависят, так что результат будет опять же тем, что нужно.

Главное в том, что два компьютера работали в параллель и следовательно — в два раза быстрее, чем один из них (если не считать потери времени на передачу данных от одного к другому).

Преждевременный вывод

Фуф! Итак MapReduce — он нужен для того, чтобы считать что-то, что либо нужно делать быстрее, либо на что не хватает памяти (либо и то, и то).

Более интересный пример — сортировка по популярности (каскады)

Допустим мы хотим посчитать количество слов в Википедии и одновременно построить список в обратном порядке их популярности — от самых популярных, к самым непопулярным.

Ясно, что все слова википедии не влезут в память, да и для обратной сортировки потом этот массив гигантский не влезет в память. Нам понадобится каскад MapReduce’ов — результат работы первого MapReduce будет поступать на вход второго MapReduce.

Если честно — я не знаю является ли слово «каскад» правильным, применимо именно к MapReduce. Это слово я использую для себя, потому что оно как никакое другое объясняет что надо сделать (результат одного водопада слов падает в MapReduce и каскадом перетекает сразу же во второй MapReduce).

Ладно, как посчитать слова — мы уже знаем:

Написанный нами шаг Map выдает:
foo, 1
bar, 1
baz, 1
foo, 1

Дальше MapReduce объединяет (сам, не Вы, как программист) их в:
bar, (1)
baz, (1)
foo, (1,1)

А написанный нами шаг Reduce выдает:
bar, 1
baz, 1
foo, 2

Теперь представим, что считали мы всю Википедию и этот массив содержит миллиарды и миллиарды слов. Отсортировать его в памяти не получится. Возмем еще один MapReduce, в этот раз Map будет делать такой фокус:

Почему минус перед цифрами?

Потому что MapReduce отсортирует всегда по возрастанию, а нам надо по убыванию. Как же используя сортировку только по возрастанию отсортировать числа в порядке уменьшения? Умножить на минус один перед сортировкой и опять на минус один после.

На вход Reduce придет такая штука:

-120, (слово3)
-15, (слово, слово2)

Прелесть, но у нас два слова имели «частоту» 15 и их сгруппировал MergeSort. Будем исправлять.

На самом деле, в зависимости от того какое воплощение MapReduce будете использовать — Вам может и не удастся в шаге Reduce выдать два массива, потому что будет требоваться только один массив на выходе — тогда просто после шага Reduce это делайте уже в своей программе.

120, слово3
15, слово,
15, слово2
1, слово4

Красота! То, что было надо.

Опять же помните, что главное что мы тут обходили — это то, что пример-то из четырех строк, а в Википедии миллиарды слов, которые в память не влезают.

Как сделать простейший MapReduce чтобы поиграться?

В коде я указал что и где должно быть, чтобы сделать более-менее полноценный MapReduce с блекдж… в смысле с файлами и MergeSort’ом. Однако это так сказать эталонная реализация, который позволит поигаться и понять как MapReduce работает. Это все еще MapReduce, просто конкретно эта реализация с точки зрения памяти ничем не выгоднее обычного хеша.

Я выбрал PHP, хотя он не самый разумный для этих целей, потому что прочитать PHP может почти любой программист, а перевести на нужный язык — это уже проще будет.

Хинты и читы

Да, в файлы я рекомендую сохранять построчно JSON представление массивов (json_encode) — меньше будет проблем с пробелами в словах, с юникодом, числами и типами данных, то есть так:
[«foo», 1]
[«bar», 1]
[«foo», 1]

То есть чтобы соединить 10 файлов с JSON представлениями достаточно:

В PHP с реализацией MergeSort, подозреваю надо возиться в полсотни строк. Если конечно в комментах никто лучше варианта не подскажет.

В Python yield и __iter__ для MapReduce — позволят сделать очень интересные вещи! Например такие:

class MapReduce — придется написать Вам самим (Я уложился в 24 строки в простейшем работающем виде, можно и меньше — упростив iter_group — это аналог функции group_tuples_by_first_element из примера для PHP).

Воплощение MapReduce на Hadoop — самое популярное решение. (Я его не пробовал, просто знаю что самое популярное).

Послесловие

Так что вот, надеюсь мой рассказ о «MapReduce без зауми» придется полезным.

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

Если будут силы — в следующем топике опишу как с помощью MapReduce считать более интересные задачки, чем банальные слова — как, например, считать ссылки в интернете, частоту слов в контексте других слов, людей в городах, товары по прайсам сотен фирм и т.п.

Да, всем ахтунг сюда! MapReduce запатентован Google, но скорее в защитных целях — тому же Hadoop они официально разрешили использовать этот метод. Так что — handle with care.

(когда-нибудь я научусь объяснять кратко. )

Источник

MapReduce 2.0. Какой он современный цифровой слон?

Если ты ИТшник, то нельзя просто так взять и выйти на работу 2-го января: пересмотреть 3-ий сезон битвы экстрасенсов или запись программы «Гордон» на НТВ (дело умственных способностей вкуса).
Нельзя потому, что у других сотрудников обязательно будут для тебя подарки: у секретарши закончился кофе, у МП — закончились дедлайны, а у администратора баз данных — амнезия память.
Оказалось, что инженеры из команды Hadoop тоже любят побаловать друг друга новогодними сюрпризами.

2 января. Упуская подробное описание эмоционально-психологического состояния лиц, участвующих в описанных ниже событиях, сразу перейду к факту: поставлен таск MAPREDUCE-279 «Map-Reduce 2.0». Оставив шутки про число, обращу внимание, что до 1-ой стабильной версии Hadoop остается чуть менее 4 лет.

За это время проект Hadoop пройдет эволюцию из маленького инновационного снежка, запущенного в 2005, в большой снежный com ком, надвигающийся на ИТ, в 2012.
Ниже мы предпримем попытку разобраться, какое же значение январский таск MAPREDUCE-279 играл (и, уверен, еще сыграет в 2013) в эволюции платформы Hadoop.

…release 1.0.0 available. After six years of gestation, Hadoop reaches 1.0.0!

и ссылку стабильную версию Hadoop v1.0.

Hadoop 2.0.0-alpha стал доступен для скачивания в конце мая. В мае же в печать вышла книга «Hadoop: The Definitive Guide, Third Edition» (автор Tom White), где довольно значительный объем отводится YARN. В начале июня Tom White выступил с презентацией «MapReduce 2.0» (видео) на Chicago Hadoop User Group. В это же месяце Cloudera с анонсировала поддержку Hadoop 2.0.0 Alpha в своем продукте CDH4. Немногим позже о поддержке Hadoop 2.0 в своих дистрибутивах заявила и компания Hortonworks.

17 сентября на сайте Apache Software Foundation было опубликовано, что YARN and MapReduce v2 доступны в Hadoop 0.23.3.

Ниже будут рассмотрены походы к распределенным вычислениям в классическом Hadoop MapReduce и новой архитектуре, описаны приемы и компоненты, реализующие концепции новой модели, а также проведено сравнение классической и 2.0 архитектур.

1. Hadoop MapReduce Classic

Hadoop – популярная программная платформа (software framework) построения распределенных приложений для массово-параллельной обработки (massive parallel processing, MPP) данных.

4K вычислительных узлов;

2. Hadoop MapReduce Next

Основные изменения коснулись компонента выполнения распределенных вычислений Hadoop MapReduce.

Классический Hadoop MapReduce представлял собой один процесс JobTracker и произвольное количество процессов TaskTracker.

ResourceManager

ResourceManager (RM) – глобальный менеджер ресурсов, чьей задачей является распределение ресурсов, затребованных приложениями и наблюдение за вычислительными узлами, на которых эти приложения выполняются.

Ресурсы у RM запрашиваются для абстрактного понятия Container, о котором речь еще пойдет и которому можно задать такие параметры как требуемое процессорное время, объем оперативной памяти, необходимая пропускная способность сети. На декабрь 2012 года поддерживается только параметр «объем RAM».

Введение RM позволяет относиться к узлам кластера как к вычислительным ресурсам, что качественно повышает утилизацию ресурсов кластера.

ApplicationMaster

ApplicationMaster (AM) – компонент, ответственный за планирование жизненного цикла, координацию и отслеживание статуса выполнения распределенного приложения. Каждое приложение имеет свой экземпляр ApplicationMaster.

На этом уровне как раз стоит рассмотреть YARN.

YARN (Yet Another Resource Negotiator) – это программный фреймворк выполнения распределенных приложений (каким экземпляр ApplicationMaster и является). YARN предоставляет компоненты и API, необходимые для разработки распределенных приложений различных типов. Сам фреймворк берет на себя ответственность по распределению ресурсов в ответ на запросы ресурсов от выполняемых приложений и ответственность за отслеживанием статуса выполнения приложений.

Модель YARN более общая (generic), чем модель, реализованная в классическом Hadoop MapReduce.

Благодаря YARN на Hadoop-кластере возможно запускать не только «map/reduce»-приложения, но и распределенные приложения, созданные с использованием: Open MPI, Spark, Apache HAMA, Apache Giraph, etc. Есть возможность реализовать и другие распределенные алгоритмы (вот она сила ООП!). Подробные инструкции описаны в Apache Wiki.

В свою очередь, MapReduce 2.0 (или MR2, или MRv2) – это фреймворк выполнения распределенных вычислений в рамках программной модели map/reduce, «лежащий» над уровнем YARN.

Разделение ответственности по управлению ресурсами и планированию/координации жизненного цикла приложения между компонентами ResourceManager и ApplicationMaster придали платформе Hadoop более распределенный характер. Что, в свою очередь, положительно сказалось на масштабируемости платформы.

NodeManager

Протоколы взаимодействия

3. Hadoop MapReduce. Vis-à-vis

В части 1 «Hadoop MapReduce Classic» было дано введение в платформу Hadoop и описаны основные ограничения платформы. В части 2 «Hadoop MapReduce Next» были описаны концепции и компоненты, введенные в новую версию фреймворка распределенных вычислений Hadoop MapReduce.

Обсудим, как концепции YARN, MR2 и компоненты, реализующие эти концепции, изменили архитектуру распределенного вычисления на платформе Hadoop, а также как эти изменения помогли (или нет) обойти с существующие ограничения платформы.

Архитектура

В Hadoop MapReduce 1.0 кластер имеет единственный узел JobTracker, который занимается распределением задач по многочисленным узлам TaskTracker, непосредственно выполняющим задачи.

В новой архитектуре Hadoop MapReduce ответственность по управлению ресурсами и планированию/координации за жизненным циклом выполнения приложений разделены между ResourceManager (per-cluster) и ApplicationMaster (per-application), соответственно.

Каждый вычислительный узел разделен на произвольное количество контейнеров Container, содержащих предопределенное количество ресурсов: CPU, RAM и т.д. Наблюдение за контейнерами ведет NodeManager (per-node).

Ниже представлена иллюстрация взаимодействия отдельных компонент Hadoop MapReduce в классическом варианте архитектуры

и YARN-подобной архитектуру (новые типы коммуникаций между компонентами выделены жирным).

Далее рассмотрим, как новая архитектура Hadoop MapReduce повлияла на такие аспекты платформы как доступность, масштабируемость, утилизация ресурсов.

Доступность

В Hadoop MapReduce 1.0 сбой JobTracker приводит к необходимости перезапуска JobTracker с чтением состояния из специальных журналов, что, в конечном итоге, приводит к простою кластера.

В новой версии решения по доступности хоть и не поднялись на качественно новый уровень, но все же дела обстоят не хуже. Hadoop MapReduce 2.0 задача высокой доступности решается следующим способом: сохраняется состояние компонентов ResourceManager и ApplicationMaster и обеспечивается система автоматического перезапуска перечисленных компонентов при сбое с подгрузкой последнего успешно сохраненного состояния.

Для ResourceManager сохранением состояния занимается Apache ZooKeeper. И при сбое менеджера ресурсов, создается новый процесс RM с состоянием, которое было до сбоя. Таким образом, последствия от сбоя RM сводятся к тому, что перезапустятся все запланированные и запущенные приложения.

Для ApplicationMaster используется собственный механизм checkpoint’ов. В процессе работы AM сохраняет свое состояние в HDFS. Если AM становится недоступным, то RM перезапускает его с состоянием из snapshot’а.

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

Разработчики, работающие с Hadoop MapReduce 1.0, неоднократно указывали, что предел масштабируемости Hadoop-кластера лежит в районе 4K машин. Основная причина этого ограничения — узел JobTracker довольно значительное количество своих ресурсов тратит на задачи, связанные с жизненным циклом приложения. Последние можно отнести к задачам специфическим для конкретного приложения, а не для кластера в целом.

Разделение ответственности за задачи, относящиеся к разным уровням, между ResourceManager и ApplicationMaster стало, пожалуй, главным ноухау Hadoop MapReduce 2.0.

Планируется, что Hadoop MapReduce 2.0 может работать на кластерах до 10K+ вычислительных узлов, что является существенным прогрессом, в сравнении с классической версией Hadoop MapReduce.

Утилизация ресурсов

Невысокая утилизация ресурсов вследствие жесткого деления ресурсов кластера на map- и reduce-слоты нередко также является объектом критики классического Hadoop MapReduce. На смену концепции слотов в MapReduce 1.0 пришла концепция универсальных контейнеров – набора взаимозаменяемых изолированных ресурсов.

Введения понятия «Container» в Hadoop MapReduce 2.0, по сути, добавил платформе Hadoop еще одно свойство – мультитенантность. Отношение к узлам кластера как к вычислительным ресурсам позволит избавиться от негативного влияния слотов на утилизацию ресурсов.

Связанность

Одной из архитектурных проблем Hadoop MapReduce 1.0 было сильная связанность 2-ух, по сути, не взаимозависимых систем: фреймворка распределенных вычислений и клиентских библиотек, реализующих распределенный алгоритм.

Это связанность стала причиной невозможности запуска на Hadoop-кластере MPI или других, альтернативных map/reduce, распределенных алгоритмов.

В новой архитектуре был выделен фреймворк распределенных вычислений YARN и фреймворк вычислений в рамках программной модели map/reduce, базирующийся на основе YARN – MR2.

MR2 является application-specific фреймворком, представленным ApplicationMaster, в то время как YARN «представлен» компонентами ResourceManager и NodeManager и полностью независим от специфики распределенного алгоритма.

За кадром

Целостной картины не будет, если не упомянуть 2 аспекта:
1. В статье рассматривался только фреймворк распределенных вычислений.
За рамками статьи остались изменения, коснувшиеся хранилища данных. Наиболее заметные из них — высокая доступность узла имен HDFS и федерации узлов имен HDFS.
2. Описанное выше будет реализовано только в Hadoop v2.0 (на время написания статьи доступен alfa-версия). Так YARN и MR2 доступны уже в Hadoop v0.23, но без поддержки высокой доступности NameNode.

Отдельно отмечу, что на июньской конференции Chicago HUG 2012, о которой я упомянул во введении, Tom White говорил, что в Hadoop 2.0 Alpha еще есть работы, связанные и с производительность, и с безопасностью, и с ResourceManager.

Заключение

Проект Hadoop в 2010 приятно удивлял идеями, в 2011 – скоростью распространения, в 2012 поразил масштабом изменений.

Не буду тратить Ваше время на «традиционное» краткое изложение того, что изменили YARN и MR2 в платформе Hadoop. Это без сомнения качественный скачок платформы.

Сейчас Hadoop выглядит как дефакто отраслевой стандарт в задачах, связанных с Big Data. Будущий релиз версии 2.0 даст разработчикам открытый, отказоустойчивый, великолепно масштабируемый, расширяемый инструмент массово-параллельной обработки, не «зацикленный» исключительно на программной модели map/reduce.

Звучит невероятно. Еще невероятнее, что это совсем недалекая реальность. Остается только один ньанс — быть этой реальности готовым.

Список источников

[1] Apache Hadoop NextGen MapReduce (YARN). Apache Software Foundation, 2011.
[2] Arun C Murthy. The Next Generation of Apache Hadoop MapReduce. Yahoo, 2011.
[3] Ahmed Radwan. MapReduce 2.0 in Hadoop 0.23. Cloudera, 2012.
[4] Tom White. Hadoop: The Definitive Guide, 3rd Edition. O’Reilly Media / Yahoo Press, 2012.
[5] Apache Hadoop Main 2.0.2-alpha API. Apache Software Foundation, 2012.

Источник