MapReduce を使用して共分散および関連量を計算する

この例では、mapreduce を使用して大規模なデータセット内のいくつかの変数の平均と共分散を計算する方法を示します。次に、共分散を使用していくつかの追跡計算を実行しますが、このときデータセット全体に対して繰り返す必要はありません。

データの準備

airlinesmall.csv データセットを使用してデータストアを作成します。この 12 MB のデータセットには、到着時間と出発時間を含む、いくつかの航空会社のフライト情報が 29 列に含まれます。この例では、ActualElapsedTime (総飛行時間)、Distance (総飛行距離)、DepDelay (フライトの出発遅延時間) および ArrDelay (フライトの到着遅延時間) を、対象の変数として選択します。

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

データストアは、既定では 'NA' 値を欠損として扱い、欠損値を NaN 値に置換します。さらに、SelectedVariableNames プロパティにより、選択した目的の変数のみを処理することができ、preview を使用して検査できます。

preview(ds)

ans=8×4 table
    ActualElapsedTime    Distance    DepDelay    ArrDelay
    _________________    ________    ________    ________

            53             308          12           8   
            63             296           1           8   
            83             480          20          21   
            59             296          12          13   
            77             373          -1           4   
            61             308          63          59   
            84             447          -2           3   
           155             954          -1          11

mapreduce の実行

関数 mapreduce は、入力として map 関数と reduce 関数を必要とします。マッパーはデータのブロックを受け取って中間結果を出力します。リデューサーは中間結果を読み取って最終結果を生成します。

次の例では、マッパーはデータストア ds 内のデータの各ブロックで、変数のカウント、平均、および共分散を計算します。次に、マッパーはブロックごとの計算値を、3 つの計算値を含む cell 配列と単一のキーからなる中間のキーと値のペアとして保存します。

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

function covarianceMapper(t,~,intermKVStore)
  % Get data from input table and remove any rows with missing values
  x = t{:,:};
  x = x(~any(isnan(x),2),:);

  % Compute and save the count, mean, and covariance
  n = size(x,1);
  m = mean(x,1);
  c = cov(x,1);

  % Store values as a single item in the intermediate key/value store
  add(intermKVStore,'key',{n m c})
end

リデューサーは各ブロックの中間結果を結合して、対象とする各変数のカウント、平均、および共分散をデータセット全体で求めます。リデューサーは、キー 'count'、'mean' および 'cov' とそれぞれの変数に対応する値の、最終的なキーと値のペアを保存します。

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

function covarianceReducer(~,intermValIter,outKVStore)
  % We will combine results computed in the mapper for different chunks of
  % the data, updating the count, mean, and covariance each time we add a new
  % chunk.

  % First, initialize everything to zero (scalar 0 is okay)
  n1 = 0; % no rows so far
  m1 = 0; % mean so far
  c1 = 0; % covariance so far

  while hasnext(intermValIter)
    % Get the next chunk, and extract the count, mean, and covariance
    t = getnext(intermValIter);
    n2 = t{1};
    m2 = t{2};
    c2 = t{3};
    
    % Use weighting formulas to update the values so far
    n = n1+n2;                     % new count
    m = (n1*m1 + n2*m2) / n;       % new mean
    
    % New covariance is a weighted combination of the two covariance, plus
    % additional terms that relate to the difference in means
    c1 = (n1*c1 + n2*c2 + n1*(m1-m)'*(m1-m) + n2*(m2-m)'*(m2-m))/ n;
    
    % Store the new mean and count for the next iteration
    m1 = m;
    n1 = n;
  end

  % Save results in the output key/value store
  add(outKVStore,'count',n1);
  add(outKVStore,'mean',m1);
  add(outKVStore,'cov',c1);
end

mapreduce を使用して、map 関数および reduce 関数をデータストア ds に適用します。

outds = mapreduce(ds, @covarianceMapper, @covarianceReducer);

********************************
*      MAPREDUCE PROGRESS      *
********************************
Map   0% Reduce   0%
Map  16% Reduce   0%
Map  32% Reduce   0%
Map  48% Reduce   0%
Map  65% Reduce   0%
Map  81% Reduce   0%
Map  97% Reduce   0%
Map 100% Reduce   0%
Map 100% Reduce 100%

mapreduce は、現在のフォルダー内のファイルでデータストア outds を返します。

関数 readall を出力データストアに使用して、mapreduce 呼び出しの結果を表示します。

results = readall(outds)

results=3×2 table
       Key                      Value               
    _________    ___________________________________

    {'count'}    {[                         120664]}
    {'mean' }    {[120.2452 703.3926 8.1334 7.1235]}
    {'cov'  }    {4x4 double                       }

Count = results.Value{1};
MeanVal = results.Value{2};
Covariance = results.Value{3};

相関行列の計算

共分散、平均およびカウントの各値は、さらに計算を実行するときに役立ちます。標準偏差を求めて、相関形式に正規化することによって、相関行列を計算します。

s = sqrt(diag(Covariance));
Correlation = Covariance ./ (s*s')

Correlation = 4×4

    1.0000    0.9666    0.0278    0.0902
    0.9666    1.0000    0.0216    0.0013
    0.0278    0.0216    1.0000    0.8748
    0.0902    0.0013    0.8748    1.0000

経過時間 (最初の列) と距離 (2 番目の列) は、Correlation(2,1) = 0.9666 であり高い相関があります。出発遅延時間 (3 番目の列) と到着遅延時間 (4 番目の列) も、Correlation(4,3) = 0.8748 で高い相関があります。

回帰係数の計算

他の 3 つの変数を予測子として使用して、到着遅延時間 ArrDelay を予測するための回帰係数を計算します。

slopes = Covariance(1:3,1:3)\Covariance(1:3,4);
intercept = MeanVal(4) - MeanVal(1:3)*slopes;
b = table([intercept; slopes], 'VariableNames', {'Estimate'}, ...
    'RowNames', {'Intercept','ActualElapsedTime','Distance','DepDelay'})

b=4×1 table
                         Estimate 
                         _________

    Intercept              -19.912
    ActualElapsedTime      0.56278
    Distance             -0.068721
    DepDelay               0.94689

PCA の実行

svd を使用して PCA (主成分分析) を実行します。主成分分析は、データセットの低次元の要約を求めるためのテクニックです。次の計算は、PCA を簡潔にしたものですが、Statistics and Machine Learning Toolbox™ の関数 pca と関数 pcacov ではより多くのオプションが使用できます。

主成分分析は、共分散または相関のいずれかを使用して実行できます。この場合、変数のスケールの差が大きいため相関を使用します。最初の 2 つの成分で分散のほとんどを捉えています。

[~,latent,pcacoef] = svd(Correlation);
latent = diag(latent)

latent = 4×1

    2.0052
    1.8376
    0.1407
    0.0164

係数行列を表示します。係数行列の各列は、ある成分が標準化された元の変数の線形結合としてどのように定義できるかを示します。最初の成分は、ほとんど最初の 2 つの変数の平均で、その他の変数からの寄与もいくらかあります。同様に、2 番目の成分はほとんどが最後の 2 つの変数の平均です。

pcacoef

pcacoef = 4×4

   -0.6291    0.3222   -0.2444   -0.6638
   -0.6125    0.3548    0.2591    0.6572
   -0.3313   -0.6244    0.6673   -0.2348
   -0.3455   -0.6168   -0.6541    0.2689

ローカル関数

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

function covarianceMapper(t,~,intermKVStore)
  % Get data from input table and remove any rows with missing values
  x = t{:,:};
  x = x(~any(isnan(x),2),:);

  % Compute and save the count, mean, and covariance
  n = size(x,1);
  m = mean(x,1);
  c = cov(x,1);

  % Store values as a single item in the intermediate key/value store
  add(intermKVStore,'key',{n m c})
end
%------------------------------------------------------------------
function covarianceReducer(~,intermValIter,outKVStore)
  % We will combine results computed in the mapper for different chunks of
  % the data, updating the count, mean, and covariance each time we add a new
  % chunk.

  % First, initialize everything to zero (scalar 0 is okay)
  n1 = 0; % no rows so far
  m1 = 0; % mean so far
  c1 = 0; % covariance so far

  while hasnext(intermValIter)
    % Get the next chunk, and extract the count, mean, and covariance
    t = getnext(intermValIter);
    n2 = t{1};
    m2 = t{2};
    c2 = t{3};
    
    % Use weighting formulas to update the values so far
    n = n1+n2;                     % new count
    m = (n1*m1 + n2*m2) / n;       % new mean
    
    % New covariance is a weighted combination of the two covariance, plus
    % additional terms that relate to the difference in means
    c1 = (n1*c1 + n2*c2 + n1*(m1-m)'*(m1-m) + n2*(m2-m)'*(m2-m))/ n;
    
    % Store the new mean and count for the next iteration
    m1 = m;
    n1 = n;
  end

  % Save results in the output key/value store
  add(outKVStore,'count',n1);
  add(outKVStore,'mean',m1);
  add(outKVStore,'cov',c1);
end
%------------------------------------------------------------------

参考

mapreduce | tabularTextDatastore