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

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

ds = tabularTextDatastore('airlinesmall.csv', 'TreatAsMissing', 'NA');
ds.ReadSize = 1000;
ds.SelectedVariableNames = {'ArrDelay', 'DepDelay', 'Distance'};

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

preview(ds)

ans=8×3 table
    ArrDelay    DepDelay    Distance
    ________    ________    ________

        8          12         308   
        8           1         296   
       21          20         480   
       13          12         296   
        4          -1         373   
       59          63         308   
        3          -2         447   
       11          -1         954   

Цепь MapReduce вызовов

Реализация мультиитерации алгоритма TSQR должна чередовать последовательные mapreduce вызовы. Чтобы продемонстрировать общий шаблон проекта цепи, этот пример использует два mapreduce итераций. Вывод от вызовов функции map передается в большой набор редукторов, и затем выход этих редукторов становится входом для следующего mapreduce итерация.

Первая итерация MapReduce

В первой итерации функция map, tsqrMapper, получает один блок (i-й) данных, который является таблицей размера $$N_i\times 3$$. mapper вычисляет $$R$$ матрица этого блока данных и сохраняет ее как промежуточный результат. Затем, mapreduce агрегирует промежуточные результаты по уникальному ключу перед отправкой их в функцию reduce. Таким образом, mapreduce отправляет все промежуточные $$R$$ матрицы с тем же ключом к тому же редуктору.

Поскольку редуктор использует qr, которая является функцией MATLAB в памяти, лучше всего сначала убедиться, что $$R$$ матрицы помещаются в памяти. Этот пример разделяет набор данных на восемь разделов. The mapreduce функция читает данные в блоках и передает данные вместе с некоторой метаинформацией в функцию map. The info входной параметр является вторым входом в функцию map, и он содержит информацию о смещении чтения и размере файла, которые необходимы для генерации ключа.

  key = ceil(offset/fileSize/numPartitions).

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

function tsqrMapper(data, info, intermKVStore)
  x = data{:,:};
  x(any(isnan(x),2),:) = [];% Remove missing values
  [~, r] = qr(x,0);

  % intermKey = randi(4); % random integer key for partitioning intermediate results
  intermKey = computeKey(info, 8);
  add(intermKVStore,intermKey, r);

  function key = computeKey(info, numPartitions)
    fileSize = info.FileSize; % total size of the underlying data file
    partitionSize = fileSize/numPartitions; % size in bytes of each partition
    offset = info.Offset; % offset in bytes of the current read
    key = ceil(offset/partitionSize);
  end
end

Функция reduce получает список промежуточных $$R$$ матрицы, вертикально конкатенирует их и вычисляет $$R$$ матрица конкатенированной матрицы.

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

function tsqrReducer(intermKey, intermValIter, outKVStore)
  x = [];

  while (intermValIter.hasnext)
    x = [x;intermValIter.getnext];
  end
  % Note that this approach assumes the concatenated intermediate values fit
  % in memory. Consider increasing the number of reduce tasks (increasing the
  % number of partitions in the tsqrMapper) and adding more iterations if it
  % does not fit in memory.

  [~, r] =qr(x,0);

  add(outKVStore,intermKey,r);
end

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

outds1 = mapreduce(ds, @tsqrMapper, @tsqrReducer);

********************************
*      MAPREDUCE PROGRESS      *
********************************
Map   0% Reduce   0%
Map  10% Reduce   0%
Map  20% Reduce   0%
Map  30% Reduce   0%
Map  40% Reduce   0%
Map  50% Reduce   0%
Map  60% Reduce   0%
Map  70% Reduce   0%
Map  80% Reduce   0%
Map  90% Reduce   0%
Map 100% Reduce   0%
Map 100% Reduce  11%
Map 100% Reduce  22%
Map 100% Reduce  33%
Map 100% Reduce  44%
Map 100% Reduce  56%
Map 100% Reduce  67%
Map 100% Reduce  78%
Map 100% Reduce  89%
Map 100% Reduce 100%

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

Вторая итерация MapReduce

Вторая итерация использует выход первой итерации outds1, как его вход. Эта итерация использует сопоставитель тождеств, identityMapper, который просто копирует данные с помощью одного ключа, 'Identity'.

Отобразите файл сопоставителя тождеств.

function identityMapper(data, info, intermKVStore)
  % This mapper function simply copies the data and add them to the
  % intermKVStore as intermediate values.
  x = data.Value{:,:};
  add(intermKVStore,'Identity', x);
end

Функция редуктора одинаковая в обеих итерациях. Использование одной клавиши функцией map означает, что mapreduce вызывает функцию сокращения только один раз во второй итерации.

Использование mapreduce применить преобразователь тождеств и тот же редуктор к выходу из первого mapreduce вызов.

outds2 = mapreduce(outds1, @identityMapper, @tsqrReducer);

********************************
*      MAPREDUCE PROGRESS      *
********************************
Map   0% Reduce   0%
Map  11% Reduce   0%
Map  22% Reduce   0%
Map  33% Reduce   0%
Map  44% Reduce   0%
Map  55% Reduce   0%
Map  66% Reduce   0%
Map  77% Reduce   0%
Map  88% Reduce   0%
Map 100% Reduce   0%
Map 100% Reduce 100%

Посмотреть Результаты

Прочтите конечные результаты из выхода datastore.

r = readall(outds2);
r.Value{:}

ans = 3×3
105 ×

    0.1091    0.0893    0.5564
         0   -0.0478   -0.4890
         0         0    3.0130

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

Вот карта и сокращение функций, которые mapreduce применяется к данным.

function tsqrMapper(data, info, intermKVStore)
  x = data{:,:};
  x(any(isnan(x),2),:) = [];% Remove missing values
  [~, r] = qr(x,0);

  % intermKey = randi(4); % random integer key for partitioning intermediate results
  intermKey = computeKey(info, 8);
  add(intermKVStore,intermKey, r);

  function key = computeKey(info, numPartitions)
    fileSize = info.FileSize; % total size of the underlying data file
    partitionSize = fileSize/numPartitions; % size in bytes of each partition
    offset = info.Offset; % offset in bytes of the current read
    key = ceil(offset/partitionSize);
  end
end
%-------------------------------------------------------------------------------
function tsqrReducer(intermKey, intermValIter, outKVStore)
  x = [];

  while (intermValIter.hasnext)
    x = [x;intermValIter.getnext];
  end
  % Note that this approach assumes the concatenated intermediate values fit
  % in memory. Consider increasing the number of reduce tasks (increasing the
  % number of partitions in the tsqrMapper) and adding more iterations if it
  % does not fit in memory.

  [~, r] =qr(x,0);

  add(outKVStore,intermKey,r);
end
%-------------------------------------------------------------------------------
function identityMapper(data, info, intermKVStore)
  % This mapper function simply copies the data and add them to the
  % intermKVStore as intermediate values.
  x = data.Value{:,:};
  add(intermKVStore,'Identity', x);
end
%-------------------------------------------------------------------------------

Ссылка