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

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

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

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

preview(ds)

ans=8×1 table
    ArrDelay
    ________

        8   
        8   
       21   
       13   
        4   
       59   
        3   
       11   

Запустить MapReduce

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

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

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

function maxArrivalDelayMapper (data, info, intermKVStore)
  partMax = max(data.ArrDelay);
  add(intermKVStore, 'PartialMaxArrivalDelay',partMax);
end

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

Просмотрите файл функции сокращения.

function maxArrivalDelayReducer(intermKey, intermValIter, outKVStore)
  % intermKey is 'PartialMaxArrivalDelay'. intermValIter is an iterator of
  % all values that has the key 'PartialMaxArrivalDelay'.
  maxVal = -Inf;
  while hasnext(intermValIter)
    maxVal = max(getnext(intermValIter), maxVal);
  end
  % The key-value pair added to outKVStore will become the output of mapreduce 
  add(outKVStore,'MaxArrivalDelay',maxVal);
end

Использовать mapreduce для применения карты и сокращения функций к хранилищу данных, ds.

maxDelay = mapreduce(ds, @maxArrivalDelayMapper, @maxArrivalDelayReducer);

********************************
*      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 100%

mapreduce возвращает хранилище данных, maxDelay, с файлами в текущей папке.

Прочтите окончательный результат из хранилища выходных данных, maxDelay.

readall(maxDelay)

ans=1×2 table
            Key             Value  
    ___________________    ________

    {'MaxArrivalDelay'}    {[1014]}

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

Здесь перечислены функции карты и сокращения, которые mapreduce относится к данным.

function maxArrivalDelayMapper (data, info, intermKVStore)
  partMax = max(data.ArrDelay);
  add(intermKVStore, 'PartialMaxArrivalDelay',partMax);
end
%--------------------------------------------------------------------
function maxArrivalDelayReducer(intermKey, intermValIter, outKVStore)
  % intermKey is 'PartialMaxArrivalDelay'. intermValIter is an iterator of
  % all values that has the key 'PartialMaxArrivalDelay'.
  maxVal = -Inf;
  while hasnext(intermValIter)
    maxVal = max(getnext(intermValIter), maxVal);
  end
  % The key-value pair added to outKVStore will become the output of mapreduce 
  add(outKVStore,'MaxArrivalDelay',maxVal);
end
%--------------------------------------------------------------------