MapReduce を使用してグループごとの要約統計量を計算する

ライブスクリプトを開く

この例では、mapreduce を使用して、グループごとに整理された要約統計量を計算する方法を示します。無名関数を使用して、パラメーター化された map 関数に追加のグループ化パラメーターを渡す方法も示します。このパラメーター化により、別のグループ化変数を使用して迅速に統計を再計算できます。

データの準備

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

ds = tabularTextDatastore('airlinesmall.csv', 'TreatAsMissing', 'NA');
ds.SelectedVariableNames = {'Month', 'UniqueCarrier', 'ArrDelay'};

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

preview(ds)

ans=8×3 table
    Month    UniqueCarrier    ArrDelay
    _____    _____________    ________

     10         {'PS'}            8   
     10         {'PS'}            8   
     10         {'PS'}           21   
     10         {'PS'}           13   
     10         {'PS'}            4   
     10         {'PS'}           59   
     10         {'PS'}            3   
     10         {'PS'}           11

mapreduce の実行

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

次の例では、マッパーはデータのブロックごとにグループ化された統計を計算し、統計を中間のキーと値のペアとして保存します。それぞれの中間のキーと値のペアには、グループレベルに対するキーと対応する統計をもつ値の cell 配列があります。

この map 関数は 4 つの入力引数を受け入れますが、関数 mapreduce では map 関数が厳密に 3 つの入力引数を受け入れなければなりません。mapreduce の呼び出し (下記) では、この追加のパラメーターを渡す方法を示します。

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

function statsByGroupMapper(data, ~, intermKVStore, groupVarName)
  % Data is a n-by-3 table. Remove missing values first
  delays = data.ArrDelay;
  groups = data.(groupVarName);
  notNaN =~isnan(delays);
  groups = groups(notNaN);
  delays = delays(notNaN);

  % Find the unique group levels in this chunk
  [intermKeys,~,idx] = unique(groups, 'stable');

  % Group delays by idx and apply @grpstatsfun function to each group
  intermVals = accumarray(idx,delays,size(intermKeys),@grpstatsfun);
  addmulti(intermKVStore,intermKeys,intermVals);

  function out = grpstatsfun(x)
    n = length(x); % count
    m = sum(x)/n; % mean
    v = sum((x-m).^2)/n; % variance
    s = sum((x-m).^3)/n; % skewness without normalization
    k = sum((x-m).^4)/n; % kurtosis without normalization
    out = {[n, m, v, s, k]};
  end
end

Map フェーズ後、mapreduce は中間のキーと値のペアを一意なキー (この場合は航空会社 ID) でグループ化するため、reduce 関数の 1 回の呼び出しでは 1 つの航空会社に関連する値が処理されます。リデューサーは、入力キー (intermKey) で指定される航空会社の中間的な統計のリストを受け取り、統計を独立したベクトル n、m、v、s および k に結合します。次に、リデューサーはこれらのベクトルを使用して、単独の航空会社のカウント、平均、分散、傾斜および尖度を計算します。最終的なキーは航空会社のコードで、関連付けられる値は 5 つのフィールドをもつ構造体に保存されます。

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

function statsByGroupReducer(intermKey, intermValIter, outKVStore)
  n = [];
  m = [];
  v = [];
  s = [];
  k = [];

  % Get all sets of intermediate statistics
  while hasnext(intermValIter)
    value = getnext(intermValIter);
    n = [n; value(1)];
    m = [m; value(2)];
    v = [v; value(3)];
    s = [s; value(4)];
    k = [k; value(5)];
  end
  % Note that this approach assumes the concatenated intermediate values fit
  % in memory. Refer to the reducer function, covarianceReducer,  of the
  % CovarianceMapReduceExample for an alternative pairwise reduction approach

  % Combine the intermediate results
  count = sum(n);
  meanVal = sum(n.*m)/count;
  d = m - meanVal;
  variance = (sum(n.*v) + sum(n.*d.^2))/count;
  skewnessVal = (sum(n.*s) + sum(n.*d.*(3*v + d.^2)))./(count*variance^(1.5));
  kurtosisVal = (sum(n.*k) + sum(n.*d.*(4*s + 6.*v.*d +d.^3)))./(count*variance^2);

  outValue = struct('Count',count, 'Mean',meanVal, 'Variance',variance,...
                 'Skewness',skewnessVal, 'Kurtosis',kurtosisVal);

  % Add results to the output datastore
  add(outKVStore,intermKey,outValue);
end

mapreduce を使用して、map 関数および reduce 関数をデータストア ds に適用します。パラメーター化された map 関数は 4 つの入力を受け入れるため、無名関数を使用して航空会社 ID を 4 番目の入力として渡します。

outds1 = mapreduce(ds, ...
    @(data,info,kvs)statsByGroupMapper(data,info,kvs,'UniqueCarrier'), ...
    @statsByGroupReducer);

********************************
*      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  10%
Map 100% Reduce  21%
Map 100% Reduce  31%
Map 100% Reduce  41%
Map 100% Reduce  52%
Map 100% Reduce  62%
Map 100% Reduce  72%
Map 100% Reduce  83%
Map 100% Reduce  93%
Map 100% Reduce 100%

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

出力データストアから最終結果を読み取ります。

r1 = readall(outds1)

r1=29×2 table
       Key           Value    
    __________    ____________

    {'PS'    }    {1x1 struct}
    {'TW'    }    {1x1 struct}
    {'UA'    }    {1x1 struct}
    {'WN'    }    {1x1 struct}
    {'EA'    }    {1x1 struct}
    {'HP'    }    {1x1 struct}
    {'NW'    }    {1x1 struct}
    {'PA (1)'}    {1x1 struct}
    {'PI'    }    {1x1 struct}
    {'CO'    }    {1x1 struct}
    {'DL'    }    {1x1 struct}
    {'AA'    }    {1x1 struct}
    {'US'    }    {1x1 struct}
    {'AS'    }    {1x1 struct}
    {'ML (1)'}    {1x1 struct}
    {'AQ'    }    {1x1 struct}
      ⋮

結果の整理

結果をより整理するために、統計を含む構造体を table に変換し、航空会社 ID を行の名前として使用します。mapreduce は、reduce 関数によってキーと値のペアが追加されたのと同じ順序でキーと値のペアを返すため、table を航空会社 ID で並べ替えます。

statsByCarrier = struct2table(cell2mat(r1.Value), 'RowNames', r1.Key);
statsByCarrier = sortrows(statsByCarrier, 'RowNames')

statsByCarrier=29×5 table
              Count     Mean      Variance    Skewness    Kurtosis
              _____    _______    ________    ________    ________

    9E          507     5.3669     1889.5      6.2676      61.706 
    AA        14578     6.9598       1123      6.0321      93.085 
    AQ          153     1.0065     230.02      3.9905      28.383 
    AS         2826     8.0771        717      3.6547      24.083 
    B6          793     11.936     2087.4      4.0072       27.45 
    CO         7999      7.048     1053.8      4.6601      41.038 
    DH          673      7.575     1491.7      2.9929      15.461 
    DL        16284     7.4971     697.48      4.4746      41.115 
    EA          875     8.2434     1221.3      5.2955      43.518 
    EV         1655     10.028     1325.4      2.9347      14.878 
    F9          332     8.4849     1138.6      4.2983      30.742 
    FL         1248     9.5144     1360.4      3.6277      21.866 
    HA          271    -1.5387     323.27      8.4245      109.63 
    HP         3597     7.5897     744.51      5.2534      50.004 
    ML (1)       69    0.15942     169.32      2.8354      16.559 
    MQ         3805     8.8591     1530.5       7.054      105.51 
      ⋮

グループ化パラメーターの変更

無名関数を使用してグループ化変数を渡すことにより、別のグループ化で統計を迅速に再計算できます。

次の例では、変数 Month を無名関数に渡すだけで、統計を再計算して結果を航空会社 ID ではなく Month でグループ化します。

outds2 = mapreduce(ds, ...
    @(data,info,kvs)statsByGroupMapper(data,info,kvs,'Month'), ...
    @statsByGroupReducer);

********************************
*      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  17%
Map 100% Reduce  33%
Map 100% Reduce  50%
Map 100% Reduce  67%
Map 100% Reduce  83%
Map 100% Reduce 100%

最終結果を読み取り、テーブルに整理します。

r2 = readall(outds2);
r2 = sortrows(r2,'Key');
statsByMonth = struct2table(cell2mat(r2.Value));
mon = {'Jan','Feb','Mar','Apr','May','Jun', ...
       'Jul','Aug','Sep','Oct','Nov','Dec'};
statsByMonth.Properties.RowNames = mon

statsByMonth=12×5 table
           Count     Mean     Variance    Skewness    Kurtosis
           _____    ______    ________    ________    ________

    Jan     9870    8.5954     973.69      4.1142      35.152 
    Feb     9160    7.3275     911.14      4.7241       45.03 
    Mar    10219    7.5536     976.34      5.1678      63.155 
    Apr     9949    6.0081     1077.4      8.9506      170.52 
    May    10180    5.2949     737.09      4.0535      30.069 
    Jun    10045    10.264     1266.1      4.8777        43.5 
    Jul    10340    8.7797     1069.7      5.1428      64.896 
    Aug    10470    7.4522     908.64      4.1959       29.66 
    Sep     9691    3.6308     664.22      4.6573      38.964 
    Oct    10590    4.6059     684.94      5.6407      74.805 
    Nov    10071    5.2835     808.65      8.0297      186.68 
    Dec    10281    10.571     1087.6      3.8564      28.823

参考

mapreduce | tabularTextDatastore

MapReduce を使用してグループごとの要約統計量を計算する

データの準備

mapreduce の実行

結果の整理

グループ化パラメーターの変更

参考

関連するトピック