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

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

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

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

preview(ds)

ans=8×1 table
    ArrDelay
    ________

        8   
        8   
       21   
       13   
        4   
       59   
        3   
       11   

Запуск MapReduce

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

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

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

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

Редуктор получает список максимальных задержек прибытия для каждого блока и находит общую максимальную задержку прибытия из списка значений. mapreduce вызывает этот редуктор только один раз, поскольку mapper добавляет только один уникальный ключ. Редуктор использует 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 для применения карты и сокращения функций к datastore, 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 возвращает datastore, maxDelay, с файлами в текущей папке.

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