Документация

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

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

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

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

preview(ds)

ans =

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

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

  key = ceil(offset/fileSize/numPartitions).

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

function tsqrMapper(data, info, intermKVStore)
% Mapper function for the TSQRMapReduceExample.

% Copyright 2014 The MathWorks, Inc.

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)
% Helper function to generate a key for the tsqrMapper function.

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);

Уменьшать функция получает список промежуточных матриц, вертикально конкатенирует их и вычисляет матрицу конкатенированной матрицы.

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

function tsqrReducer(intermKey, intermValIter, outKVStore)
% Reducer function for the TSQRMapReduceExample.

% Copyright 2014 The MathWorks, Inc.

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);

outKVStore.add(intermKey,r);

Используйте 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)
% Mapper function for the MapReduce TSQR example.
%
% This mapper function simply copies the data and add them to the
% intermKVStore as intermediate values.

% Copyright 2014 The MathWorks, Inc.

x = data.Value{:,:};
add(intermKVStore,'Identity', x);

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

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

function tsqrReducer(intermKey, intermValIter, outKVStore)
% Reducer function for the TSQRMapReduceExample.

% Copyright 2014 The MathWorks, Inc.

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);

outKVStore.add(intermKey,r);

Используйте 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 =

   1.0e+05 *

    0.1091    0.0893    0.5564
         0   -0.0478   -0.4890
         0         0    3.0130

Ссылка