Big Data: MapReduce

3.1 История появление термина BigData

Термин Big Data появился сравнительно недавно. Google Trends показывает начало активного роста употребления словосочетания начиная с 2011 года:

При этом уже сейчас термин не использует только ленивый. Особенно часто не по делу термин используют маркетологи. Так что же такое Big Data на самом деле? Раз уж я решил системно изложить и осветить вопрос – необходимо определиться с понятием.

В своей практике я встречался с разными определениями:

• Big Data – это когда данных больше, чем 100Гб (500Гб, 1ТБ, кому что нравится).
• Big Data – это такие данные, которые невозможно обрабатывать в Excel.
• Big Data – это такие данные, которые невозможно обработать на одном компьютере.

И даже такие:

• Вig Data – это вообще любые данные.
• Big Data не существует, ее придумали маркетологи.

Я буду придерживаться определения с wikipedia:

Большие данные (англ. big data) — серия подходов, инструментов и методов обработки структурированных и неструктурированных данных огромных объёмов и значительного многообразия для получения воспринимаемых человеком результатов, эффективных в условиях непрерывного прироста, распределения по многочисленным узлам вычислительной сети, сформировавшихся в конце 2000-х годов, альтернативных традиционным системам управления базами данных и решениям класса Business Intelligence.

Таким образом под Big Data я буду понимать не какой-то конкретный объём данных и даже не сами данные, а методы их обработки, которые позволяют распредёлено обрабатывать информацию. Эти методы можно применить как к огромным массивам данных (таким как содержание всех страниц в интернете), так и к маленьким (таким как содержимое этой лекции).

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

• Логи поведения пользователей в интернете
• GPS-сигналы от автомобилей для транспортной компании
• Данные, снимаемые с датчиков в большом адронном коллайдере
• Оцифрованные книги в Российской Государственной Библиотеке
• Информация о транзакциях всех клиентов банка
• Информация обо всех покупках в крупной ритейл сети и т.д.

Количество источников данных стремительно растёт, а значит технологии их обработки становятся всё более востребованными.

3.2 Принципы работы с большими данными

Исходя из определения Big Data, можно сформулировать основные принципы работы с такими данными:

1. Горизонтальная масштабируемость. Поскольку данных может быть сколь угодно много – любая система, которая подразумевает обработку больших данных, должна быть расширяемой. В 2 раза вырос объём данных – в 2 раза увеличили количество железа в кластере и всё продолжило работать.

2. Отказоустойчивость. Принцип горизонтальной масштабируемости подразумевает, что машин в кластере может быть много. Например, Hadoop-кластер Yahoo имеет более 42000 машин (по этой ссылке можно посмотреть размеры кластера в разных организациях). Это означает, что часть этих машин будет гарантированно выходить из строя. Методы работы с большими данными должны учитывать возможность таких сбоев и переживать их без каких-либо значимых последствий.

3. Локальность данных. В больших распределённых системах данные распределены по большому количеству машин. Если данные физически находятся на одном сервере, а обрабатываются на другом – расходы на передачу данных могут превысить расходы на саму обработку. Поэтому одним из важнейших принципов проектирования BigData-решений является принцип локальности данных – по возможности обрабатываем данные на той же машине, на которой их храним.

Все современные средства работы с большими данными так или иначе следуют этим трём принципам. Для того, чтобы им следовать – необходимо придумывать какие-то методы, способы и парадигмы разработки средств разработки данных. Один из самых классических методов я разберу в сегодняшней лекции.

3.3 MapReduce

MapReduce – это модель распределенной обработки данных, предложенная компанией Google для обработки больших объёмов данных на компьютерных кластерах. MapReduce неплохо иллюстрируется следующей картинкой:

MapReduce предполагает, что данные организованы в виде некоторых записей. Обработка данных происходит в 3 стадии:

1. Стадия Map. На этой стадии данные предобрабатываются при помощи функции map(), которую определяет пользователь. Работа этой стадии заключается в предобработке и фильтрации данных. Работа очень похожа на операцию map в функциональных языках программирования – пользовательская функция применяется к каждой входной записи.

Функция map() примененная к одной входной записи и выдаёт множество пар ключ-значение. Множество – т. е. может выдать только одну запись, может не выдать ничего, а может выдать несколько пар ключ-значение. Что будет находиться в ключе и в значении – решать пользователю, но ключ – очень важная вещь, так как данные с одним ключом в будущем попадут в один экземпляр функции reduce.

2. Стадия Shuffle. Проходит незаметно для пользователя. В этой стадии вывод функции map «разбирается по корзинам» – каждая корзина соответствует одному ключу вывода стадии map. В дальнейшем эти корзины послужат входом для reduce.

3. Стадия Reduce. Каждая «корзина» со значениями, сформированная на стадии shuffle, попадает на вход функции reduce().

Функция reduce задаётся пользователем и вычисляет финальный результат для отдельной «корзины». Множество всех значений, возвращённых функцией reduce(), является финальным результатом MapReduce-задачи.

Несколько дополнительных фактов про MapReduce:

1. Все запуски функции map работают независимо и могут работать параллельно, в том числе на разных машинах кластера.
2. Все запуски функции reduce работают независимо и могут работать параллельно, в том числе на разных машинах кластера.
3. Shuffle внутри себя представляет параллельную сортировку, поэтому также может работать на разных машинах кластера. Пункты 1-3 позволяют выполнить принцип горизонтальной масштабируемости.
4. Функция map, как правило, применяется на той же машине, на которой хранятся данные – это позволяет снизить передачу данных по сети (принцип локальности данных).
5. MapReduce – это всегда полное сканирование данных, никаких индексов нет. Это означает, что MapReduce плохо применим, когда ответ требуется очень быстро.

3.4 Примеры задач, эффективно решаемых при помощи MapReduce

Word Count

Начнём с классической задачи – Word Count. Задача формулируется следующим образом: имеется большой корпус документов. Задача – для каждого слова, хотя бы один раз встречающегося в корпусе, посчитать суммарное количество раз, которое оно встретилось в корпусе.

Решение:

Раз имеем большой корпус документов – пусть один документ будет одной входной записью для MapRreduce–задачи. В MapReduce мы можем только задавать пользовательские функции, что мы и сделаем (будем использовать python-like псевдокод):

def map(doc): 
for word in doc: 
yield word, 1 

def reduce(word, values): 
yield word, sum(values) 

Функция map превращает входной документ в набор пар (слово, 1), shuffle прозрачно для нас превращает это в пары (слово, [1,1,1,1,1,1]), reduce суммирует эти единички, возвращая финальный ответ для слова.

Обработка логов рекламной системы

Второй пример взят из реальной практики Data-Centric Alliance.

Задача: имеется csv-лог рекламной системы вида:

<user_id>,<country>,<city>,<campaign_id>,<creative_id>,<payment></p> 
 
11111,RU,Moscow,2,4,0.3 
22222,RU,Voronezh,2,3,0.2 
13413,UA,Kyiv,4,11,0.7 
… 

Необходимо рассчитать среднюю стоимость показа рекламы по городам России.

Решение:

def map(record): 
user_id, country, city, campaign_id, creative_id, payment = record.split(",") 
payment=float(payment) 
if country == "RU": 
yield city, payment 

def reduce(city, payments): 
yield city, sum(payments)/len(payments) 

Функция map проверяет, нужна ли нам данная запись – и если нужна, оставляет только нужную информацию (город и размер платежа). Функция reduce вычисляет финальный ответ по городу, имея список всех платежей в этом городе.