Blogerator.org

× Версия для печати

Уж целую неделю мы постепенно разжевываем NoSQL/NewSQL, но вопросы не убывают (как я ожидал), но только нарастают.

Разделавшись намедни с основами команд memcached, сейчас я хочу попытаться максимально просто, насколько это возможно для меня, ответить на частый и важный вопрос — что такое MapReduce.

Что такое MapReduce?

Это типовой подход, алгоритм, ну или паттерн, тут уж как кто назовет, параллельной обработки больших объемов сырых данных, например — результатов работы краулеров или логов веб-запросов.

Вообще по статистике, до 80% задач могут свободно и очень выгодно маппиться на MapReduce, и именно MapReduce драйвит сейчас в NoSQL.

Основы основ

Итак, типичная реализация этого алгоритма получает на вход 3 аргумента: исходную коллекцию, Map-функцию, Reduce-функцию, — и возвращает новую коллекцию данных после обработки.

Collection MapReduce(Collection source, Function map, Function reduce)

Алгоритм состоит из нескольких шагов. В качестве первого шага выполняется Map-функция к каждому элементу исходной коллекции. Map вернет ноль либо создаст экземпляры Key/Value объектов.

ArrayOfKeyValue Map(object itemFromSourceCollection)

То есть, можно сказать, что обязанность Map-функции конвертировать элементы исходной коллекции в ноль или несколько экземпляров Key/Value объектов. Это продемонстрировано ниже на изображении:

Следующим шагом, алгоритм отсортирует все пары Key/Value и создаст новые экземпляры объектов, где все значения (value ) будут сгруппированы по ключу.

Последним шагом выполнится функция Reduce — для каждого сгруппированного экземпляра Key/Value объекта

ItemResult Reduce(KeyWithArrayOfValues item)

В заключении, функция Reduce вернет новый экземпляр объекта, который будет включен в результирующую коллекцию.

Пример-реализация

В качестве примера реализуем очень простую и наглядную имплементацию этого алгоритма на C#. Мой пример считает количество гласных построчно в наборе строк.

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

Итак, поехали:

using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading.Tasks;
 
namespace MapReduceSample
{
   // Элемент результирующей коллекции, гласная и ее количество
    class VocalCount
    {
        public char Vocal;
        public int Count;
    }
 
    clаss Program
    {
        stаtic vоid Mаin(string[] args)
        {
            // "lines" - это исходная коллекция.
            var lines = new[] {
                "How many vocals do",
                "these two lines have?"
            };
 
            forеach (var line in lines)
            {
                Cоnsole.WriteLine(line);
            }
            Cоnsole.WriteLine();
 
            // Вызывается MapReduce
            var results = MаpReduce(lines, Map, Reduce);
 
            // Отображение результата
            foreаch (var result in results)
            {
                Consоle.WriteLine("{0} = {1}", result.Vocal, result.Count);
            }
 
            Consоle.ReadKey();
        }
 
        /// <summary>
        /// Функция Map считает количество гласных в строке
        /// </summary>
        /// <param name="sourceItem" />Строка для подсчета</param>
        /// <returns>Коллекция экземпляров Key/Value.
        /// Где key - гласная, и значение ее количество.</returns>
        static IEnumеrable<KeyValuePair<char, int>> Map(string sourceItem)
        {
            return sоurceItem
                .ToLower()
                .Where(c => "aeiou".Contains(c))
                .GroupBy(c => c, (c, instances) => new KeyValuePair<char, int>(c, instances.Count()));
 
        }
 
        /// <summary>
        /// Функция Reduce считает общее число каждой гласной
        /// </summary>
         /// <param name="reduceItem" />Экземпляр Key/Values, где key - гласная,
        /// и value - список всех подсчетов гласных по строкам</param>
        /// <returns>Экземпляр элемента результирующей коллекции, VocalCоunt</returns>
        static VocalCount Reduce(KeyValuePair<char, IEnumerable<int>> reduceItem)
        {
            return new VocalCount
            {
                Vocal = rеduceItem.Key,
                Count = rеduceItem.Value.Sum()  // Computes total count
            };
        }
 
        /// <summary>
        /// Обобщенная реализация функции MapReduce
        /// </summary>
        /// <typeparam name="TSource">Тип элементов исходной коллекции</typeparam>
        /// <typeparam name="TKey">Типа ключа Key используемого в функциях Map и Reduce</typeparam>
        /// <typeparam name="TValue">Тип Value используемый в функциях Map и Reduce</typeparam>
        /// <typeparam name="TResult">Тип элементов результирующей коллекции</typeparam>
        /// <param name="source" />Исходная коллекция</param>
        /// <param name="map" />Функция Map</param>
        /// <param name="reduce" />Функция Reduce</param>       
        static IEnumerable<TResult>  MapReduce<TSource, TKey, TValue, TResult>(
            IEnumerable<TSource> source,
            Func<TSource, IEnumerable<KeyValuePair<TKey, TValue>>> map,
            Func<KeyValuePair<TKey, IEnumerable<TValue>>, TResult> reduce)
        {
            // Коллекция, где сохраним результаты нашей map-функции
            var mapResults = new ConcurrentBag<KeyValuePair<TKey, TValue>>();
 
            // Выполним функцию Map паралельно для каждого элемента исходной коллекции
            Parallel.ForEаch(source, sourceItem =>
            {
                foreаch (var mapResult in map(sourceItem))
                {
                    mapResults.Add(mapResult);
                }
            });
 
            // Сгрупируем все значения по ключам
            var reduceSources = mapResults.GroupBy(
                    item => itеm.Key,
                    (key, values) => new KeyValuePair<TKey, IEnumеrable<TValue>>(key, values.Select(i=>i.Value)));
 
            var resultCollection = new BlockingCollection<TResult>();
 
            // Старуем reduce
            Task.Factory.StartNew(() =>
            {
                // Выполняем функцию Reduce паралельно для каждого элемента reduceSources
                Parallel.ForEаch(reduceSources,
                                (reduceItem) => resultCollection.Add(reduce(reduceItem)));
                 
                resultCollection.CompleteAdding();
            });
             
            return resultCollection.GetConsumingEnumerable();
        }
    }
}

Довесок: Хронология развития технологии MapReduce, также для всех жаждущих серьёзных подробностей могу порекомендовать очень хороший курс на эту тему «MapReduce course at the University of Maryland for Spring 2013», выложенный полностью в онлайн.