MapReduce

MapReduce — это модель программирования и связанная с ней реализация для обработки и создания больших наборов данных с помощью распределенного параллельного алгоритма в кластере .

Программа MapReduce состоит из карты процедуры , которая выполняет фильтрацию и сортировку (например, сортировку студентов по имени в очереди, по одной очереди для каждого имени), и метода сокращения , который выполняет суммарную операцию (например, подсчет количества учащихся). студентов в каждой очереди, что дает частоту имен). «Система MapReduce» (также называемая «инфраструктурой» или «фреймворком») организует обработку, объединяя распределенные серверы, параллельно выполняя различные задачи, управляя всеми коммуникациями и передачей данных между различными частями системы и обеспечивая резервирование. и отказоустойчивость .

Модель является специализацией стратегии разделения-применения-объединения для анализа данных. Он вдохновлен функциями отображения и сокращения , обычно используемыми в функциональном программировании . хотя их назначение в рамках MapReduce не такое же, как в их исходных формах. Ключевым вкладом платформы MapReduce не являются фактические функции карты и сокращения (которые, например, напоминают стандарт интерфейса передачи сообщений 1995 года ). уменьшать и разбросать операций), но масштабируемость и отказоустойчивость, достигаемые для множества приложений за счет распараллеливания. Таким образом, однопоточная реализация MapReduce обычно не быстрее, чем традиционная реализация (не MapReduce); любой выигрыш обычно наблюдается только при многопоточных реализациях на многопроцессорном оборудовании. Использование этой модели выгодно только тогда, когда в игру вступают оптимизированная операция распределенного перемешивания (которая снижает затраты на сетевую связь) и функции отказоустойчивости инфраструктуры MapReduce. Оптимизация затрат на связь важна для хорошего алгоритма MapReduce.

MapReduce Библиотеки написаны на многих языках программирования с разными уровнями оптимизации. Популярная реализация с открытым исходным кодом , поддерживающая распределенное перемешивание, является частью Apache Hadoop . Название MapReduce первоначально относилось к собственной технологии Google , но с тех пор было обобщено . К 2014 году Google больше не использовал MapReduce в качестве основной модели обработки больших данных . и разработка Apache Mahout перешла к более функциональным и менее ориентированным на диск механизмам, которые включали полную карту и возможности сокращения.

Обзор

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

Платформа (или система) MapReduce обычно состоит из трех операций (или шагов):

1. Карта: каждый рабочий узел применяет map функцию в локальные данные и записывает выходные данные во временное хранилище. Главный узел гарантирует, что обрабатывается только одна копия избыточных входных данных.
2. Перетасовка: рабочие узлы перераспределяют данные на основе выходных ключей (создаваемых map функция), так что все данные, принадлежащие одному ключу, расположены на одном рабочем узле.
3. Уменьшение: рабочие узлы теперь обрабатывают каждую группу выходных данных для каждого ключа параллельно.

MapReduce позволяет выполнять распределенную обработку карт и операций сокращения. Карты могут выполняться параллельно при условии, что каждая операция картирования независима от других; на практике это ограничено количеством независимых источников данных и/или количеством процессоров рядом с каждым источником. Аналогичным образом, набор «редукторов» может выполнять фазу сокращения при условии, что все выходные данные операции отображения, использующие один и тот же ключ, одновременно представляются одному и тому же редуктору или что функция сокращения является ассоциативной . Хотя этот процесс часто кажется неэффективным по сравнению с более последовательными алгоритмами (поскольку необходимо запускать несколько экземпляров процесса сокращения), MapReduce можно применять к значительно большим наборам данных, чем может обработать один «обычный» сервер – большая ферма серверов может использовать MapReduce для сортировки петабайта данных всего за несколько часов. Параллелизм также предлагает некоторую возможность восстановления после частичного отказа серверов или хранилища во время операции: если один маппер или редюсер выйдет из строя, работу можно перенести — при условии, что входные данные все еще доступны.

Другой способ взглянуть на MapReduce — это 5-этапное параллельное и распределенное вычисление:

1. Подготовьте ввод Map() — «система MapReduce» назначает процессоры Map, назначает входной ключ K1 , с которым будет работать каждый процессор, и предоставляет этому процессору все входные данные, связанные с этим ключом.
2. Запустите предоставленный пользователем код Map() — Map() запускается ровно один раз для каждого ключа K1 , генерируя выходные данные, организованные по ключу K2 .
3. «Перетасуйте» выходные данные Map на процессоры сокращения — система MapReduce назначает процессоры сокращения, назначает ключ K2 , с которым должен работать каждый процессор, и предоставляет этому процессору все сгенерированные Map данные, связанные с этим ключом.
4. Запустите предоставленный пользователем код Редуцировать() — Редукция() запускается ровно один раз для каждого ключа K2 , созданного на этапе Карты.
5. Создание окончательного результата — система MapReduce собирает все выходные данные сокращения и сортирует их по K2 для получения окончательного результата.

Логически эти пять шагов можно рассматривать как выполняемые последовательно – каждый шаг начинается только после завершения предыдущего шага – хотя на практике их можно чередовать, если это не повлияет на конечный результат.

Во многих ситуациях входные данные могли быть уже распределены ( «сегментированы» ) между множеством различных серверов, и в этом случае шаг 1 иногда можно значительно упростить, назначив картографические серверы, которые будут обрабатывать локально присутствующие входные данные. Аналогичным образом, шаг 3 иногда можно ускорить, назначив процессоры сокращения, которые максимально приближены к данным, сгенерированным Map, которые им необходимо обработать.

Логическое представление

MapReduce и . MapReduce функции Обе определены в отношении данных, структурированных в парах (ключ, значение) Карта принимает одну пару данных с типом в одном домене данных и возвращает список пар в другом домене:

Map(k1,v1) → list(k2,v2)

Функция Map применяется параллельно к каждой паре (под ключом k1) во входном наборе данных. Это создает список пар (с ключами k2) для каждого звонка. После этого фреймворк MapReduce собирает все пары с одним и тем же ключом ( k2) из всех списков и группирует их вместе, создавая по одной группе для каждого ключа.

Затем функция уменьшения применяется параллельно к каждой группе, что, в свою очередь, создает коллекцию значений в одном домене:

Reduce(k2, list (v2)) → list((k3, v3))

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

Таким образом, платформа MapReduce преобразует список пар (ключ, значение) в другой список пар (ключ, значение). Это поведение отличается от типичной комбинации карты функционального программирования и сокращения, которая принимает список произвольных значений и возвращает одно единственное значение, объединяющее все значения, возвращаемые картой.

Для реализации MapReduce необходимо , но недостаточно иметь реализации карты и сократить абстракции. Распределенные реализации MapReduce требуют средств соединения процессов, выполняющих этапы Map и сокращения. Это может быть распределенная файловая система . Возможны и другие варианты, такие как прямая потоковая передача от преобразователей к преобразователям или передача результатов процессорами сопоставления преобразователям, которые их запрашивают.

Примеры

Канонический пример MapReduce подсчитывает появление каждого слова в наборе документов:

function map(String name, String document):
    // name: document name
    // document: document contents
    for each word w in document:
        emit (w, 1)

function reduce(String word, Iterator partialCounts):
    // word: a word
    // partialCounts: a list of aggregated partial counts
    sum = 0
    for each pc in partialCounts:
        sum += pc
    emit (word, sum)

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

В качестве другого примера представьте, что для базы данных из 1,1 миллиарда человек нужно вычислить среднее количество социальных контактов, которые имеет человек в зависимости от возраста. В SQL такой запрос может быть выражен как:

  SELECT age, AVG(contacts)
    FROM social.person
GROUP BY age
ORDER BY age

Используя MapReduce, Значениями ключа K1 могут быть целые числа от 1 до 1100, каждое из которых представляет пакет из 1 миллиона записей. Ключевое значение K2 может представлять собой возраст человека в годах, и это вычисление может быть достигнуто с использованием следующих функций:

function Map is
    input: integer K1 between 1 and 1100, representing a batch of 1 million social.person records
    for each social.person record in the K1 batch do
        let Y be the person's age
        let N be the number of contacts the person has
        produce one output record (Y,(N,1))
    repeat
end function

function Reduce is
    input: age (in years) Y
    for each input record (Y,(N,C)) do
        Accumulate in S the sum of N*C
        Accumulate in Cnew the sum of C
    repeat
    let A be S/Cnew
    produce one output record (Y,(A,Cnew))
end function

Обратите внимание, что в уменьшения , Функция C — количество людей, имеющих в общей сложности N контактов, поэтому в Функцию карты естественно написать C=1 , поскольку каждая выходная пара относится к контактам одного человека.

Система MapReduce выстроит в ряд 1100 процессоров Map и предоставит каждому соответствующий 1 миллион входных записей. Шаг Map даст 1,1 миллиарда (Y,(N,1)) записи, причем Значения Y находятся в диапазоне, скажем, от 8 до 103. Затем система MapReduce выстроит в ряд 96 процессоров сокращения, выполняя операцию перетасовки пар ключ/значение, поскольку нам нужно среднее значение по возрасту, и предоставит каждому свои миллионы соответствующие входные записи. Шаг «Сокращение» приведет к значительному сокращению набора выходных записей, состоящего всего из 96 записей. (Y,A) , который будет помещен в окончательный файл результатов, отсортированный по И .

Информация о количестве в записи важна, если обработка сокращается более одного раза. Если бы мы не добавили количество записей, вычисленное среднее значение было бы неправильным, например:

-- map output #1: age, quantity of contacts
10, 9
10, 9
10, 9

-- map output #2: age, quantity of contacts
10, 9
10, 9

-- map output #3: age, quantity of contacts
10, 10

Если мы уменьшим файлы №1 и #2 у нас будет новый файл со средним числом 9 контактов для 10-летнего человека ((9+9+9+9+9)/5):

-- reduce step #1: age, average of contacts
10, 9

Если мы уменьшим его с помощью файла #3 , мы теряем счет того, сколько записей мы уже просмотрели, поэтому в итоге у нас получается в среднем 9,5 контактов для 10-летнего человека ((9+10)/2), что неверно. Правильный ответ: 9,1 66 = 55/6 = (9×3+9×2+10×1)/(3+2+1).

Поток данных

Архитектура программной платформы придерживается принципа открытости-закрытости , при котором код эффективно разделяется на неизменяемые « замороженные» точки и расширяемые « горячие точки» . Замороженное место фреймворка MapReduce — это большой распределенный тип. Горячие точки, которые определяет приложение:

• считыватель ввода
• карты ​ функция
• функция распределения ​
• функция сравнения ​
• Функция уменьшения ​
• писатель выходной

Считыватель ввода

Устройство чтения входных данных делит входные данные на «разделения» соответствующего размера (на практике обычно от 64 МБ до 128 МБ), и платформа назначает одно разделение для каждой функции карты . Устройство чтения ввода считывает данные из стабильного хранилища (обычно распределенной файловой системы ) и генерирует пары ключ/значение.

Типичный пример: чтение каталога, полного текстовых файлов, и возврат каждой строки в виде записи.

Функция карты

Функция Map принимает серию пар ключ/значение, обрабатывает каждую и генерирует ноль или более выходных пар ключ/значение. Типы ввода и вывода карты могут отличаться (и часто отличаются) друг от друга.

Если приложение выполняет подсчет слов, функция карты разбивает строку на слова и выводит пару ключ/значение для каждого слова. Каждая выходная пара будет содержать слово в качестве ключа и количество экземпляров этого слова в строке в качестве значения.

Функция разделения

Каждый вывод функции Map назначается конкретному редуктору приложения с помощью функции секционирования для целей сегментирования . Функция секционирования получает ключ и количество редукторов и возвращает индекс нужного редуктора .

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

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

Функция сравнения

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

Функция сокращения

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

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

Писатель вывода

Модуль записи вывода записывает выходные данные сокращения в стабильное хранилище.

Теоретическая основа

Свойства моноидов являются основой для обеспечения корректности операций MapReduce.

В пакете Algebird реализация Map/Reduce в Scala явно требует типа класса monoid .

Операции MapReduce имеют дело с двумя типами: тип A отображаемых входных данных и тип B сокращаемых выходных данных.

Операция Map принимает отдельные значения типа A и создает для каждого a:A значение b:B ; Для операции сокращения требуется бинарная операция, • определенная для значений типа B ; он состоит в сведении всех доступных значений b:B к одному значению.

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

• ассоциативность: ( x • y ) • z = x • ( y • z )
• существование нейтрального элемента e такого, что e • x = x • e = x для каждого x:B .

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

Эти два свойства сводятся к наличию моноида ( B , •, e ) для значений типа B с операцией • и нейтральным элементом e .

Нет никаких требований к значениям типа A ; произвольную функцию A → B. можно использовать Map для операции Это означает, что мы имеем катаморфизм A* → ( B , •, e ). Здесь A* обозначает звезду Клини также известную как тип списков над A. ,

Операция Shuffle сама по себе не имеет отношения к сути MapReduce; необходимо распределить вычисления по облаку.

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

• построение дерева из поддеревьев: эта операция не ассоциативна, и результат будет зависеть от группировки;
• прямой расчет средних: avg также не ассоциативен (и в нем нет нейтрального элемента); чтобы вычислить среднее значение, нужно вычислить моменты .

производительности

Программы MapReduce не гарантированно будут быстрыми. Основное преимущество этой модели программирования заключается в использовании оптимизированной операции перемешивания платформы и необходимости писать только Map и уменьшить части программы . Однако на практике автору программы MapReduce приходится учитывать этап перемешивания; в частности, функция секционирования и объем данных, записываемых функцией Map , могут иметь большое влияние на производительность и масштабируемость. Дополнительные модули, такие как функция «Объединитель», могут помочь уменьшить объем данных, записываемых на диск и передаваемых по сети. Приложения MapReduce могут добиться сублинейного ускорения при определенных обстоятельствах.

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

При настройке производительности MapReduce необходимо учитывать сложность отображения, перемешивания, сортировки (группировки по ключу) и сокращения. Объем данных, создаваемых картографами, является ключевым параметром, который перемещает большую часть вычислительных затрат между картографированием и сокращением. Сокращение включает в себя сортировку (группировку ключей), имеющую нелинейную сложность. Следовательно, небольшие размеры разделов сокращают время сортировки, но здесь есть компромисс, поскольку наличие большого количества редукторов может быть непрактичным. Влияние размера разделенного блока незначительно (если только он не выбран особенно неудачно, скажем, <1 МБ). Выигрыш от чтения нагрузки с локальных дисков некоторыми картографами в среднем незначителен.

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

и

MapReduce обеспечивает надежность за счет разделения ряда операций над набором данных на каждый узел сети. Ожидается, что каждый узел будет периодически отчитываться о выполненной работе и обновлениях статуса. Если узел молчат дольше этого интервала, главный узел (аналогично главному серверу в файловой системе Google ) записывает узел как неработающий и отправляет назначенную ему работу другим узлам. Отдельные операции используют атомарные операции для именования выходных файлов в качестве проверки, чтобы убедиться в отсутствии запущенных параллельных конфликтующих потоков. Когда файлы переименовываются, их можно также скопировать под другое имя помимо имени задачи (с учетом побочных эффектов ).

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

Реализации не обязательно обладают высокой надежностью. Например, в старых версиях был единственной Hadoop NameNode точкой отказа распределенной файловой системы. Более поздние версии Hadoop имеют высокую доступность с активным/пассивным переключением при отказе для «NameNode».

Использует

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

В Google MapReduce использовалась для полной регенерации индекса Google во Всемирной паутине . Он заменил старые специальные программы, которые обновляли индекс и проводили различные анализы. С тех пор разработка в Google перешла на такие технологии, как Percolator, FlumeJava. и MillWheel , которые предлагают потоковую обработку и обновления вместо пакетной обработки, что позволяет интегрировать «живые» результаты поиска без перестроения полного индекса.

Стабильные входные и выходные данные MapReduce обычно хранятся в распределенной файловой системе . Временные данные обычно хранятся на локальном диске и извлекаются редукторами удаленно.

Критика

Отсутствие новизны

Дэвид ДеВитт и Майкл Стоунбрейкер , ученые-компьютерщики, специализирующиеся на параллельных базах данных и архитектурах без разделяемого доступа , критически оценили широту проблем, для решения которых можно использовать MapReduce. Они назвали его интерфейс слишком низкоуровневым и поставили под сомнение, действительно ли он представляет собой сдвиг парадигмы, о котором заявляют его сторонники. Они оспорили утверждения сторонников MapReduce о новизне, сославшись на Teradata как на пример предшествующего уровня техники , существующего уже более двух десятилетий. Они также сравнили программистов MapReduce с программистами CODASYL , отметив, что оба «пишут на низкоуровневом языке , выполняя низкоуровневые манипуляции с записями». Использование MapReduce входных файлов и отсутствие поддержки схемы препятствует повышению производительности, обеспечиваемому обычными функциями системы баз данных, такими как B-деревья и хэш-секционирование , хотя такие проекты, как Pig (или PigLatin) , Sawzall , Apache Hive , HBase и Бигтабл решают некоторые из этих проблем.

Грег Йоргенсен написал статью, опровергающую эти взгляды. Йоргенсен утверждает, что весь анализ ДеВитта и Стоунбрейкера беспочвенен, поскольку MapReduce никогда не разрабатывалась и не предназначалась для использования в качестве базы данных.

Впоследствии ДеВитт и Стоунбрейкер опубликовали в 2009 году подробное сравнительное исследование, сравнивающее производительность подходов Hadoop MapReduce и RDBMS для решения нескольких конкретных задач. Они пришли к выводу, что реляционные базы данных предлагают реальные преимущества для многих видов использования данных, особенно при сложной обработке или в тех случаях, когда данные используются на предприятии, но что пользователям может быть проще адаптировать MapReduce для простых или одноразовых задач обработки.

Парадигма программирования MapReduce также была описана в диссертации Дэнни Хиллиса 1985 года. предназначен для использования на Connection Machine , где это называлось «xapping/reduction». и полагался на специальное оборудование этой машины для ускорения картирования и уменьшения. Диалект, который в конечном итоге использовался для Connection Machine, StarLisp 1986 года , имел параллельные *map и reduce!!, который, в свою очередь, был основан на Common Lisp 1984 года , который имел непараллельные map и reduce встроенный. Древовидный . Connection Machine архитектура гиперкуба подход, который использует для выполнения reduce в $O(\log n)$ время По сути, это то же самое, что и подход, упомянутый в документе Google как предыдущая работа.

В 2010 году Google получил так называемый патент на MapReduce. Патент, поданный в 2004 году, может охватывать использование MapReduce программным обеспечением с открытым исходным кодом, таким как Hadoop , CouchDB и другими. Редактор журнала Ars Technica признал роль Google в популяризации концепции MapReduce, но усомнился в том, является ли патент действительным или новым. В 2013 году в рамках своего «Обещания не утверждать патенты (OPN)» Google обязалась использовать патент только в защитных целях. Ожидается, что срок действия патента истечет 23 декабря 2026 года.

Ограниченная среда программирования

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

См. также

• Формализм Бёрда – Меертенса
• Контракт на распараллеливание

Реализации MapReduce

• Apache CouchDB
• Апач Хадуп
• Инфиниспан
• Пульсация