MapReduce を使用する tall QR (TSQR) 行列の因数分解

ライブスクリプトを開く

この例では、mapreduce を使用して tall 行列の QR (TSQR) の因数分解を行う方法を示します。mapreduce 呼び出しを連結して複数の因数分解を繰り返し実行する方法を示し、map 関数の info 引数を使用して数値キーを計算します。

データの準備

airlinesmall.csv データセットを使用してデータストアを作成します。この 12 MB のデータセットには、到着時間と出発時間を含む、いくつかの航空会社のフライト情報が 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 呼び出しを連結する必要があります。一般的な連結設計パターンを示すために、この例では 2 回の mapreduce 反復を使用します。map 関数の呼び出しからの出力は、リデューサーの大きなセットに渡され、これらのリデューサーの出力は次の mapreduce 反復の入力になります。

最初の MapReduce 反復

最初の反復では、map 関数 tsqrMapper はデータの 1 つのブロック (i 番目) を受け取ります。これは、サイズが $N_{i} \times 3$ の table です。マッパーは、データのこのブロックの $R$ 行列を計算し、中間結果として保存します。次に、mapreduce は、中間結果を reduce 関数に送る前に、一意なキーごとに集計します。このようにして、mapreduce は同じキーをもつ中間の $R$ 行列をすべて同じリデューサーに送ります。

リデューサーではインメモリ MATLAB® 関数である qr を使用するため、まず行列 $R$ がメモリに収まるのを確認することをお勧めします。この例では、データセットを 8 分割に分けます。関数 mapreduce はブロック内のデータを読み取り、メタ情報と一緒に map 関数に渡します。info 入力引数は map 関数への 2 番目の入力で、キーを生成するために必要な読み取りオフセットとファイルサイズ情報を含みます。

  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$ 行列を計算します。

reduce 関数のファイルを表示します。

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 関数をデータストア 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 の 2 回目の反復

2 回目の反復では、最初の反復の出力 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 は 2 回目の反復では reduce 関数を一度のみ呼び出すということです。

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
10⁵ ×

    0.1091    0.0893    0.5564
         0   -0.0478   -0.4890
         0         0    3.0130

ローカル関数

ここに挙げるのは、mapreduce がデータに適用する map 関数と reduce 関数です。

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

Reference

Paul G. Constantine and David F. Gleich. 2011. Tall and skinny QR factorizations in MapReduce architectures. In Proceedings of Second International Workshop on MapReduce and Its Applications (MapReduce '11). ACM, New York, NY, USA, 43-50.DOI=10.1145/1996092.1996103 https://doi.acm.org/10.1145/1996092.1996103

参考

mapreduce | tabularTextDatastore