![](https://habrastorage.org/getpro/habr/upload_files/79f/ce7/6b1/79fce76b1f251ed2c4763a0699333b6a.png)
Здраствуйте читатели)
Перед тем как начать, хочу отметить, что данная статья служит для любопытствующих людей и тем, кому нужно базовое теоретическое знание MapReduce'а. Сам по себе MapReduce уже устарел. Если вы в поиске хороших решений, то увы, в этой статье не будет вестись речь о готовых инструментах. О них вы можете почитать в комментариях или написать свой, буду благодарен)
(P.S: Спасибо за замечания, интерес и активность Revertis, procfg, Stas911 и другим^^)
Пример задачи
Хотим автоматизировать огромный фруктовый рынок. На каждое событие будем писать строчку в структурированный лог.
![](https://habrastorage.org/getpro/habr/upload_files/c0f/0af/3c0/c0f0af3c0ec2e0a8e0c7ac2acad25382.png)
Этот лог не является частью runtime функционирования рынка, но может быть полезен для изучения статистики и аналитики. Например, на основании лога продавец может сделать вывод, что свежие яблоки выгоднее привозить к 13:00.
Хотим узнать
Сколько всего денег покупатели потратили на каждый тип фруктов?
Почему бы не реляционная бд + SQL?
Пусть все эти данные лежат в таблице log в MySQL и мы просто делаем такой запрос:
select fruit, sum(price)
from log
where event = 'buy'
group by fruit;
Это будет работать. Но наш рынок огромный! И размер этого лога 100 петабайт. Он не поместится на жесткий диск одной машины.
А что если в Hadoop?
Пусть этот лог пишется в hadoop кластер. Hadoop сам разложит эту "таблицу" на разные серверы (обычно называется "ноды" node).
Например так:
![](https://habrastorage.org/getpro/habr/upload_files/b2e/b3a/30f/b2eb3a30f6c0eafab9e381b58e2c70ea.png)
За распределение данных по кластеру отвечает HDFS (Hadoop Distributed File System). Из-за того, что данные разделены на разные ноды, объем данных не ограничен размером жесткого диска одной ноды. А репликация обеспечивает отказоустойчивость (если одна из нод сломается, реплика встанет на ее место).
Теперь с помощью двух операций map и reduce получим ответ на наш вопрос.
Пишем map
Операция map очень похожа на функцию map из функционального программирования: она преобразовывает каждую строчку. Но в добавок она может еще и выфильтровать ненужные строчки.
У map из ФП сигнатура такая, например:
Row map(Row row)
Она принимает некий объект класса Row, который представляет строку, и возвращает другой объект Row, который может иметь совсем другой формат. Однако, чтобы такая функция могла еще и фильтровать, ее нужно немного усложнить. Добавим параметр context, в который будем писать строчку вместо того, чтобы ее возвращать (но только если это необходимо):
void map(Row row, Context context)
Теперь напишем имплементацию этой функции, которая:
1. Выбрасывает все события, кроме buy
2. Выбрасывает все столбцы, кроме fruit и price
void map(Row row, Context context) {
if (row.get("event").equals("buy")) {
context.write(new Row(
"fruit", row.get("fruit"),
"price", row.get("price")
));
}
}
Это функция будет вызываться для каждой строчки нашей таблицы. Результат map:
![](https://habrastorage.org/getpro/habr/upload_files/e32/769/d2b/e32769d2bc0599ac3a7f61beb9952cae.png)
Мы получили новую (временную) таблицу.
Теперь должна произойти магия.
Как работает Shuffle?
Этот шаг произведет сам hadoop. Мы только должны сказать ему, какой столбец будет ключом. В нашем случае ключом будет столбец fruit.
Ноды будут обмениваться строчками так, чтобы
все строчки с одинаковым ключом попали на одну ноду
Результат будет такой:
![](https://habrastorage.org/getpro/habr/upload_files/ba6/570/61a/ba657061a7c5f106c4ec9d673ed44fce.png)
Пишем reduce
Теперь, когда мы уверены, что все бананы лежат на одной и той же ноде (ровно как и другие фрукты), мы можем посчитать их стоимость.
Функция reduce, совсем не отличается от функции reduce в ФП, однако для единообразия возвращать будем так же как в map - через context.
void reduce(List<Row> rows, Context context);
Функция принимает список строчек с одинаковым ключом. И выполняется для каждой такой группы.
Реализация должна просто посчитать сумму:
void reduce(List<Row> rows, Context context) {
int sum = 0;
for (Row row : rows) {
sum += row.get("price");
}
context.write(new Row(
"fruit", rows.get(0).get("fruit"),
"price", sum
));
}
Такой получим результат:
![](https://habrastorage.org/getpro/habr/upload_files/acf/bb3/40e/acfbb340e56760f38d69dc35e8375959.png)
Если вычитать из HDFS новую (временную) таблицу целиком, то получим точно такой же результат, как и с SQL:
![](https://habrastorage.org/getpro/habr/upload_files/1bd/574/afd/1bd574afd6027ef8897fd74b3b9e9a32.png)
Выводы
С помощью MapReduce можно сделать все, что можно сделать с SQL.
MapReduce сложнее принять, но он изначально нацелен на масштабирование - в кластере могут быть десятки тысяч нод, а твой запрос будет работать почти также, как и с десятком нод.
Если данных не так много, то лучше использовать твою любимую реляционную БД, потому что это намного проще и быстрее.
P.S, Старался максимально упростить материал, поэтому в некоторых местах могут быть маленькие обманчики, которые помогают легче понять концепцию.