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

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

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

Хранилище данных обрабатывает '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, получает один блок (i-й) данных, представляющий собой таблицу размера $${}_{\mathrm{Ni}}\mathrm{\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 для применения карты и сокращения функций к хранилищу данных, 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 возвращает хранилище выходных данных, 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%

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

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

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

Ссылка