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

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

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

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

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

  key = ceil(offset/fileSize/numPartitions).

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

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

Уменьшать функция получает список промежуточного звена $$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 применять map и reduce функции к 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

Функция редуктора является тем же самым в обеих итерациях. Использование одного ключа функцией карты означает тот 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

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

Перечисленный здесь map и reduce функции что 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
%-------------------------------------------------------------------------------

Ссылка