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

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

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

Хранилище данных обрабатывает '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

После фазы «Карта» 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 для применения карты и сокращения функций к хранилищу данных, 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 возвращает хранилище данных, meanDelayByDay, с файлами в текущей папке.

Прочтите окончательный результат из хранилища выходных данных, 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   

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

Здесь перечислены функции карты и сокращения, которые 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
%---------------------------------------------------------------------------