Подготовка данных

Создайте datastore с помощью airlinesmall.csv набор данных. Этот набор данных на 12 мегабайтов содержит 29 столбцов информации о рейсе для нескольких поставщиков услуг авиакомпании, включая прибытие и время отправления. В этом примере выберите DayOfWeek и ArrDelay (задержка прибытия рейса) как переменные интереса.

ds = tabularTextDatastore('airlinesmall.csv', 'TreatAsMissing', 'NA');
ds.SelectedVariableNames = {'ArrDelay', 'DayOfWeek'};

Datastore обрабатывает 'NA' значения как пропавшие без вести и замены отсутствующие значения с NaN значения по умолчанию. Кроме того, SelectedVariableNames свойство позволяет вам работать только с выбранными переменными интереса, который можно проверить использование preview.

preview(ds)

ans=8×2 table
    ArrDelay    DayOfWeek
    ________    _________

        8           3    
        8           1    
       21           5    
       13           5    
        4           4    
       59           3    
        3           4    
       11           6    

Выполнение mapreduce

mapreduce функция требует функции карты и уменьшать функции как входные параметры. Картопостроитель получает блоки данных и выходные промежуточные результаты. Редуктор читает, промежуточное звено заканчивается и приводит к конечному результату.

В этом примере картопостроитель вычисляет количество и сумму задержек поденно недели в каждом блоке данных, и затем хранит результаты как промежуточные пары "ключ-значение". Ключи являются целыми числами (1 - 7), представление дней недели и значений является двухэлементными векторами, представляющими количество и сумму задержки каждого дня.

Отобразите файл функции карты.

function meanArrivalDelayByDayMapper(data, ~, intermKVStore)
  % Data is an n-by-2 table: first column is the DayOfWeek and the second
  % is the ArrDelay. Remove missing values first.
  delays = data.ArrDelay;
  day = data.DayOfWeek;
  notNaN = ~isnan(delays);
  day = day(notNaN);
  delays = delays(notNaN);

  % find the unique days in this chunk
  [intermKeys,~,idx] = unique(day, 'stable');

  % group delays by idx and apply @grpstatsfun function to each group
  intermVals = accumarray(idx,delays,size(intermKeys),@countsum);
  addmulti(intermKVStore,intermKeys,intermVals);

  function out = countsum(x)
    n = length(x); % count
    s = sum(x); % mean
    out = {[n, s]};
  end
end

После фазы Map, mapreduce группирует промежуточные пары "ключ-значение" уникальным ключом (в этом случае, день недели). Таким образом каждый вызов редуктора работает над значениями, сопоставленными с одним днем недели. Редуктор получает список промежуточного количества и сумму задержек в течение дня, заданного входным ключом (intermKey) и подводит итог значений в общее количество, n и полная сумма s. Затем редуктор вычисляет полное среднее значение и добавляет одну итоговую пару "ключ-значение" в выход. Эта пара "ключ-значение" представляет среднюю задержку прибытия рейса в течение одного дня недели.

Отобразите уменьшать файл функции.

function meanArrivalDelayByDayReducer(intermKey, intermValIter, outKVStore)
  n = 0;
  s = 0;

  % get all sets of intermediate results
  while hasnext(intermValIter)
    intermValue = getnext(intermValIter);
    n = n + intermValue(1);
    s = s + intermValue(2);
  end

  % accumulate the sum and count
  mean = s/n;
  % add results to the output datastore
  add(outKVStore,intermKey,mean);
end

Используйте mapreduce применять map и reduce функции к datastore, ds.

meanDelayByDay = mapreduce(ds, @meanArrivalDelayByDayMapper, ...
                               @meanArrivalDelayByDayReducer);

********************************
*      MAPREDUCE PROGRESS      *
********************************
Map   0% Reduce   0%
Map  16% Reduce   0%
Map  32% Reduce   0%
Map  48% Reduce   0%
Map  65% Reduce   0%
Map  81% Reduce   0%
Map  97% Reduce   0%
Map 100% Reduce   0%
Map 100% Reduce  14%
Map 100% Reduce  29%
Map 100% Reduce  43%
Map 100% Reduce  57%
Map 100% Reduce  71%
Map 100% Reduce  86%
Map 100% Reduce 100%

mapreduce возвращает datastore, meanDelayByDay, с файлами в текущей папке.

Считайте конечный результат из выходного datastore, meanDelayByDay.

result = readall(meanDelayByDay)

result=7×2 table
    Key      Value   
    ___    __________

     3     {[7.0038]}
     1     {[7.0833]}
     5     {[9.4193]}
     4     {[9.3185]}
     6     {[4.2095]}
     2     {[5.8569]}
     7     {[6.5241]}

Организация результатов

Целочисленные ключи (1 - 7) представляют дни недели. Чтобы организовать результаты больше, преобразуйте ключи в категориальный массив, получите числовые значения из одной ячеек элемента и переименуйте имена переменных получившейся таблицы.

result.Key = categorical(result.Key, 1:7, ...
               {'Mon','Tue','Wed','Thu','Fri','Sat','Sun'});
result.Value = cell2mat(result.Value);
result.Properties.VariableNames = {'DayOfWeek', 'MeanArrDelay'}

result=7×2 table
    DayOfWeek    MeanArrDelay
    _________    ____________

       Wed          7.0038   
       Mon          7.0833   
       Fri          9.4193   
       Thu          9.3185   
       Sat          4.2095   
       Tue          5.8569   
       Sun          6.5241   

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

result = sortrows(result,'MeanArrDelay')

result=7×2 table
    DayOfWeek    MeanArrDelay
    _________    ____________

       Sat          4.2095   
       Tue          5.8569   
       Sun          6.5241   
       Wed          7.0038   
       Mon          7.0833   
       Thu          9.3185   
       Fri          9.4193   

Локальные функции

Перечисленный здесь map и reduce функции что mapreduce применяется к данным.

function meanArrivalDelayByDayMapper(data, ~, intermKVStore)
  % Data is an n-by-2 table: first column is the DayOfWeek and the second
  % is the ArrDelay. Remove missing values first.
  delays = data.ArrDelay;
  day = data.DayOfWeek;
  notNaN = ~isnan(delays);
  day = day(notNaN);
  delays = delays(notNaN);

  % find the unique days in this chunk
  [intermKeys,~,idx] = unique(day, 'stable');

  % group delays by idx and apply @grpstatsfun function to each group
  intermVals = accumarray(idx,delays,size(intermKeys),@countsum);
  addmulti(intermKVStore,intermKeys,intermVals);

  function out = countsum(x)
    n = length(x); % count
    s = sum(x); % mean
    out = {[n, s]};
  end
end
%---------------------------------------------------------------------------
function meanArrivalDelayByDayReducer(intermKey, intermValIter, outKVStore)
  n = 0;
  s = 0;

  % get all sets of intermediate results
  while hasnext(intermValIter)
    intermValue = getnext(intermValIter);
    n = n + intermValue(1);
    s = s + intermValue(2);
  end

  % accumulate the sum and count
  mean = s/n;
  % add results to the output datastore
  add(outKVStore,intermKey,mean);
end
%---------------------------------------------------------------------------