# Scaling Distributed Machine Learning with the Parameter Server
Navigation: [[Parameter Server]] | [[Mu Li]]
> [!abstract]
> 分散機械学習問題向けのパラメータサーバフレームワークを提案する。
> データとワークロードの両方をワーカーノード群に分散し、サーバノード群は密または疎なベクトル・行列として表現される大域共有パラメータを保持する。
> このフレームワークはノード間の非同期データ通信を管理し、柔軟な一貫性モデル・弾力的なスケーラビリティ・継続的な障害耐性を提供する。
> 提案フレームワークのスケーラビリティを示すため、疎ロジスティック回帰から潜在的ディリクレ配分(Latent Dirichlet Allocation)・分散スケッチングまでの問題に対して、ペタバイト規模の実データ・数十億の事例と数十億のパラメータを用いた実験結果を示す。
## 論文情報
| 項目 | 内容 |
|------|------|
| 論文名 | Scaling Distributed Machine Learning with the Parameter Server |
| 著者 | [[Mu Li]]∗‡, [[David G. Andersen]]∗, Jun Woo Park∗, [[Alexander J. Smola]]∗†, Amr Ahmed†, Vanja Josifovski†, James Long†, Eugene J. Shekita†, Bor-Yiing Su† |
| 所属 | ∗[[Carnegie Mellon University]] / ‡[[Baidu]] / †[[Google]] |
| 会議 | OSDI 2014(11th USENIX Symposium on Operating Systems Design and Implementation) |
| 開催日時 | 2014年10月6〜8日、Broomfield, CO |
| URL | https://www.usenix.org/conference/osdi14/technical-sessions/presentation/li_mu |
## 概要
本論文は、分散機械学習(ML)向けのパラメータサーバの**第三世代**実装を提案する。ワーカーノード群がデータと計算を分担し、サーバノード群が大域共有パラメータ(疎・密なベクトル・行列として表現)を保持・同期する。既存のパラメータサーバ(第一世代: memcached ベースの key-value ストア流用、第二世代: YahooLDA・Distbelief 等の用途特化実装)の限界を、非同期通信・柔軟な一貫性モデル・弾力的スケーラビリティ・継続的障害耐性・線形代数向けデータ型という5つの特徴で克服する。疎ロジスティック回帰・LDA・分散スケッチングの3種のアルゴリズムで、既存に報告された公開実験のいずれよりも大きい規模(コア数・パラメータ数)を実証した(図1)。
**Figure 1: 公開されている最大規模の機械学習実験の比較**
![[_attachments/osdi14-parameter-server/fig01-scale-comparison.png]]
(図1. 横軸にコア数、縦軸に共有パラメータ数を取り、各システムが公開した最大規模の実験をプロットしたもの。青丸は疎ロジスティック回帰、赤四角は潜在変数グラフィカルモデル、灰色五角形は深層ネットワークを表す。パラメータサーバ(Sparse LR・LDA)は他のどのシステムよりも数桁大きい規模を達成している。)
## 問題設定
### 背景
大規模機械学習では、訓練データが1TBから1PBに達し、モデルパラメータ数も10⁹から10¹²に及ぶ。これらのパラメータは全ワーカーノードで大域的に共有され、計算のたびに頻繁にアクセスされる。共有には3つの課題がある。
1. **通信帯域**: パラメータへのアクセスには膨大なネットワーク帯域が必要になる。
2. **同期の障壁**: 多くの機械学習アルゴリズムは逐次的で、同期の障壁が性能を損なう。特にマシンレイテンシの分散が大きい環境ではその影響が顕著になる。
3. **障害耐性**: 学習タスクはクラウド環境で実行されることが多く、マシンの不安定性やジョブのプリエンプションが発生する。ある大規模インターネット企業の3か月分のジョブログでは、10,000マシン時間規模のジョブの失敗率が24.7%に達する(表1)。
### 先行研究との関係
| 世代 | 代表例 | 特徴 | 限界 |
|------|--------|------|------|
| 第一世代 | [43](Smola & Narayanamurthy, VLDB 2010) | memcached の (key,value) ストアを流用 | 柔軟性・性能が低い |
| 第二世代 | YahooLDA、Distbelief | ユーザ定義の更新プリミティブを持つ専用サーバ | 用途特化に留まる |
| 第三世代(本研究) | Petuum(境界遅延モデル) を含む一般化 | 一般プラットフォーム化 | Petuum はワーカースレッドモデルに制約を課す |
表2は既存の分散データ解析システム(Graphlab・Petuum・REEF・Naiad・Mlbase)とパラメータサーバの属性(共有データ形式・一貫性・障害耐性)を比較し、パラメータサーバが唯一「継続的障害耐性」を持つことを示す。
| システム | 共有データ | 一貫性 | 障害耐性 |
|---------|-----------|--------|---------|
| Graphlab | グラフ | eventual | チェックポイント |
| Petuum | ハッシュテーブル | delay bound | なし |
| REEF | 配列 | BSP | チェックポイント |
| Naiad | (key,value) | multiple | チェックポイント |
| Mlbase | テーブル | BSP | RDD |
| **Parameter Server** | **(疎)ベクトル/行列** | **多様** | **継続的** |
## 提案手法
### アーキテクチャ
パラメータサーバのインスタンスは、1つのサーバ群と複数のワーカー群から構成される(図4)。サーバ群の各サーバノードは大域共有パラメータの一部を保持し、サーバマネージャがメタデータ(ノードの生死・パラメータ分割の割当)の一貫したビューを維持する。ワーカー群は訓練データの一部をローカルに保持し、局所統計量(勾配など)を計算する。ワーカーはサーバノードとのみ通信し、ワーカー間の直接通信は行わない。タスクスケジューラがワーカー群ごとに存在し、タスク割当と進行監視、ワーカー増減時の再スケジューリングを担う。
**Figure 4: パラメータサーバのアーキテクチャ**
![[_attachments/osdi14-parameter-server/fig04-architecture.png]]
(図4. サーバ群(サーバマネージャ+複数サーバノード)と複数のワーカー群(タスクスケジューラ+複数ワーカーノード)の関係。リソースマネージャがサーバ群・ワーカー群の起動を管理する。)
### (Key,Value)ベクトルとしてのパラメータ表現
共有モデルは (key, value) のペア集合として表現される。既存手法との違いは、パラメータサーバがこれらを**線形代数オブジェクト**として扱う点にある。キーが順序付けられていることを前提に、ベクトル・行列の加算・乗算・ノルム計算などの最適化された演算を提供し、BLAS・LAPACK・ATLAS のような CPU 効率の良いマルチスレッド線形代数ライブラリを活用できる。
### Range Push と Pull
`w.push(R, dest)` と `w.pull(R, dest)` により、キー範囲 R に対する一括のプッシュ・プルを行う。範囲ベースの通信により、プログラマの利便性と帯域効率を両立させる。
### 分散サブグラジェント降下法の例
図2・アルゴリズム1に基本的な分散サブグラジェント降下法を示す。各ワーカーは局所勾配 gᵣ を計算してサーバへ push し、サーバは全ワーカーの勾配を集約(g = Σgᵣ)して重み w を更新し、更新後の w をワーカーへ pull させる。
**Figure 2: 分散サブグラジェント降下法の手順**
![[_attachments/osdi14-parameter-server/fig02-subgradient-descent-steps.png]]
(図2. ワーカー1〜mが局所勾配を計算(1)、サーバへpush(2)、サーバが集約・更新(3)、ワーカーがpull(4)する4ステップのループ。各ワーカーは全パラメータではなくwの作業集合のみをキャッシュする。)
パラメータ空間が大きくても、各ワーカーが実際に必要とする作業集合は小さい。100ワーカーではパラメータの7.8%、10,000ワーカーでは0.15%に縮小する(図3)。
**Figure 3: ワーカー数増加に伴うパラメータ作業集合の縮小**
![[_attachments/osdi14-parameter-server/fig03-parameters-per-worker.png]]
(図3. 横軸をワーカー数、縦軸を1ワーカーあたりのパラメータ割合(%、対数)としたグラフ。ワーカー数が増えるほど各ワーカーが保持すべきパラメータの割合が単調に減少する。)
### タスクの非同期実行と依存関係
タスクはリモート手続き呼び出し(push・pull・ユーザ定義関数)として発行され、既定では非同期に並列実行される。呼び出し元は依存関係(execute-after-finished)を設定して実行順序を制御できる。
**Figure 5: イテレーション間の依存関係**
![[_attachments/osdi14-parameter-server/fig05-iteration-dependency.png]]
(図5. イテレーション12はイテレーション11に依存するが、10と11は独立であるため非同期に処理できる例。)
### 柔軟な一貫性モデル
依存関係グラフの形により、Sequential(逐次)・Eventual(結果整合性)・Bounded Delay(境界遅延)の3種の一貫性モデルを実現できる(図6)。境界遅延 τ を0にすると逐次一貫性、∞にすると結果整合性になり、τ を調整することでアルゴリズムの収束速度とシステム効率のトレードオフを制御できる。
**Figure 6: 一貫性モデルごとの有向非巡回グラフ(DAG)**
![[_attachments/osdi14-parameter-server/fig06-consistency-dags.png]]
(図6. (a)Sequential: 完全な直列依存、(b)Eventual: 依存なし、(c)境界遅延1: 隣接するイテレーションのみ依存。遅延を大きくするほどDAGのサイズが増える。)
### ユーザ定義フィルタ
「有意に変化したフィルタ(significantly modified filter)」は閾値を超えて変化した要素のみを送信する。評価では、最適化の最適性条件(KKT条件)を利用した「KKTフィルタ」を導入し、更新の影響が小さいと推定される勾配の送信を抑制する。
### 実装: ベクトルクロック・メッセージ・一貫性ハッシュ・複製
- **ベクトルクロック(§4.1)**: 各(key,value)ペアにノードごとの時刻を記録する。ナイーブな実装ではO(nm)の空間が必要だが、範囲ベース通信のため多数のパラメータが同じタイムスタンプを共有することを利用し、範囲ベクトルクロックとして圧縮する(アルゴリズム2)。
- **メッセージ圧縮(§4.2)**: キーのキャッシュ(送信側がキーリストのハッシュのみ送る)と Snappy 圧縮ライブラリによる値の圧縮(ゼロエントリの除去)を組み合わせる。
- **一貫性ハッシュ(§4.3)**: 従来の分散ハッシュテーブル同様にキーとサーバノードIDをハッシュリング上に配置し、各サーバノードがマスタとして担当するキー範囲を持つ(図7)。
- **複製と一貫性(§4.4)**: 各サーバノードは反時計回りにk個の隣接キー範囲をスレーブとして複製する。集約後複製(replication after aggregation)により、n人のワーカーからの更新をサーバ側で集約した後に複製することで、複製に必要な帯域をk/n倍に抑える(図8)。
- **サーバ・ワーカー管理(§4.5・4.6)**: ノード追加・離脱時にサーバマネージャ・タスクスケジューラがキー範囲の再割当と未完了タスクの再送を行う。
**Figure 7・8: キーリングと複製生成**
![[_attachments/osdi14-parameter-server/fig07-08-key-ring-and-replication.png]]
(図7(左). キー範囲とサーバノードIDをハッシュリング上に配置し、各サーバが自ノードの担当範囲(owned by S1)と複製範囲(replicated by S1)を持つ様子。図8(右). 複製生成の例。単一ワーカーの場合(左)はワーカーW1がS1にxをpushし、S1がf(x)を計算してスレーブS2に複製した後にackを返す。複数ワーカーの場合(右)はS1がW1・W2からのx・yを集約(x+y)してからf(x+y)を計算・複製するため、複製に使う帯域がk/n倍に抑えられる。)
## 新規性
1. **継続的障害耐性を持つ唯一の一般プラットフォーム**: 表2に示すように、比較対象の全システムの中でパラメータサーバのみが継続的な障害耐性(1秒未満の復旧)を提供する。
2. **線形代数オブジェクトとしてのネイティブなデータ型**: 単純な(key,value)抽象化ではなく、ベクトル・行列演算を第一級でサポートする点が第一世代・第二世代のパラメータサーバとの違いである。
3. **柔軟な一貫性モデルの一般化**: Sequential・Eventual・Bounded Delay を単一のタスク依存グラフの枠組みで統一的に表現し、アルゴリズム設計者に選択の自由を与える。
4. **実装コストの大幅な削減**: System A・System B はいずれも10,000行超のコード(System Bは30,000行)を要するのに対し、パラメータサーバは同等の機能を300行で実現できる(表3)。
## 実験設定
| 項目 | 疎ロジスティック回帰 | LDA | スケッチング |
|------|---------------------|-----|-------------|
| データ規模 | 1700億件・650億特徴、636TB(非圧縮)/141TB(圧縮) | 50億ユニークユーザーID、500万ドメイン | 3000億エントリ、1億超のユニークキー |
| クラスタ規模 | 1000台(800ワーカー+200サーバ)、各16コア・192GB DRAM、10Gb Ethernet | 800ワーカー+200サーバ、および5000ワーカー+1000サーバ、各10コア・128GB DRAM、10Gb/s以上 | 15台(90仮想サーバノード)、各64コア、40Gb Ethernet |
| アルゴリズム | Delayed Block Proximal Gradient(アルゴリズム3) | Stochastic Variational Methods + Collapsed Gibbs sampling + 分散勾配降下法 | CountMin Sketch(アルゴリズム4) |
| 比較対象 | System A(L-BFGS・逐次)、System B(Block PG・逐次) | YahooLDA・Graphlab・Petuum(いずれも同規模を扱えないことを著者に確認) | (単独評価) |
## 実験結果
### 疎ロジスティック回帰
パラメータサーバは同一アルゴリズム(Block PG)を用いる System B より高速に収束し、L-BFGS を用いる System A よりもさらに高速だった(図9)。System A のワーカーは32%、System B は53%がバリア待機で idle 状態だったが、パラメータサーバでは2%未満に抑えられた(図10)。
**Figure 9・10: 収束曲線とワーカー時間の内訳**
![[_attachments/osdi14-parameter-server/fig09-convergence-sparse-lr.png]]
(図9. 目的関数値(対数軸)の時間推移。System-A・System-Bに比べパラメータサーバが最も速く低い目的関数値に到達する。)
![[_attachments/osdi14-parameter-server/fig10-time-computation-waiting.png]]
(図10. 各システムのワーカー1台あたりの計算時間(computing)と待機時間(waiting)の内訳。パラメータサーバは待機時間が最も小さい。)
キーのキャッシュとKKTフィルタによるネットワークトラフィック削減効果を図11・図12に示す。圧縮込みでサーバ側は約40倍、ワーカー側は約12.3倍のトラフィック削減を達成した。KKTフィルタは93%超のユニーク特徴量を除外する。
**Figure 11・12: トラフィック削減とKKTフィルタの効果**
![[_attachments/osdi14-parameter-server/fig11-network-traffic-savings.png]]
(図11. baseline・+caching keys・+KKT filterの各段階での相対ネットワークトラフィック(非圧縮/圧縮)。左がサーバ側で圧縮により最大40.8倍、右がワーカー側で+caching keysのみでは1.9〜2.5倍にとどまるが+KKT filterを加えると12.3倍まで削減される。)
![[_attachments/osdi14-parameter-server/fig12-filtered-features.png]]
(図12. 最適化の進行に伴い、KKTフィルタで除外されるユニーク特徴量の割合が94.5%から97.4%まで上昇する様子。)
境界遅延 τ を変化させた実験では、τ=0(逐次)でワーカーの50%が idle だったのに対し、τ=16では1.7%まで低下した。ただし計算時間は τ にほぼ線形に増加するため、τ=8がアルゴリズム収束と系効率の最良のトレードオフだった(図13)。
**Figure 13: 境界遅延τごとの時間内訳**
![[_attachments/osdi14-parameter-server/fig13-time-by-max-delay.png]]
(図13. τ=0,1,2,4,8,16それぞれについての計算時間と待機時間の内訳。τが大きいほど待機時間は減るが、τ=16では計算時間自体が増加している。)
### LDA(潜在的ディリクレ配分)
1000台から6000台にマシンを増やすと、収束が約4倍高速化した(図14)。従来報告されていた最大規模の実験([2])はアクティブユーザー1億人未満・トークン10万未満・トピック1000未満だったのに対し、本実験は50億ユーザー・500万トークン・2000トピックという2桁大きい規模を扱った。
**Figure 14: LDAの収束**
![[_attachments/osdi14-parameter-server/fig14-lda-convergence.png]]
(図14左. 1000台・50億ユーザーでのワーカーごとの対数尤度分布の時間推移。低い値はストラグラーによるもの。右: 1000台と6000台マシンでの収束時間の比較。)
学習されたトピック例として、「Programming」「Music」「Baby Related」「Strength Training」などのユーザー興味クラスタが得られた(表4)。
### スケッチング(CountMin Sketch)
15台のマシン(90仮想サーバノード)で、ピーク挿入速度13億件/秒、平均11億件/秒を達成し、サーバノード障害からの復旧を0.8秒で完了した(表5)。
| 指標 | 値 |
|------|-----|
| ピーク挿入速度 | 13億件/秒 |
| 平均挿入速度 | 11億件/秒 |
| マシンあたりピーク帯域 | 4.37 Gbit/s |
| 障害ノード復旧時間 | 0.8秒 |
## 考察
パラメータサーバは「線形代数オブジェクトとしてのパラメータ表現」「範囲ベースの非同期通信」「柔軟な一貫性モデル」という3つの設計判断の相乗効果により、疎ロジスティック回帰・LDA・スケッチングという性質の大きく異なる3種のアルゴリズムを同一フレームワークでサポートできることを実証した。著者らは、機械学習アルゴリズム側がデータの摂動に比較的頑健であることを利用し、システム制約(厳密な一貫性など)を緩和することが本研究の核心的な洞察であるとしている。
## 強み
- 疎ロジスティック回帰・LDA・スケッチングという異種のアルゴリズムを同一フレームワークで、既存の専用実装(System A・B)や他システム(YahooLDA・Graphlab・Petuum)を上回る規模で実行できる。
- キーキャッシュとメッセージ圧縮の組み合わせで最大40倍規模のネットワークトラフィック削減を達成する。
- サーバノード障害からの復旧が1秒未満と高速で、計算を中断しない。
- 専用実装(System B、30,000行)と比較して実装コードが300行に収まり、機械学習アルゴリズムの記述をシステム詳細から分離できる。
## 弱点・課題
- ベクトルクロックの空間効率化(範囲ベクトルクロックへの圧縮)は m・k が小さいことを前提としており、範囲の統合(merge)は将来課題として残されている。
- クラスタリソースマネージャ(YARN・Mesos)との統合は将来課題であり、本論文の評価では独自のノード管理機構に依存している。
- 境界遅延 τ の最適値(実験ではτ=8)はデータ・アルゴリズム・ハードウェアに依存し、事前に自動選択する機構は提供されない。
- ワーカー障害時の復旧をスキップするか再割当するかの判断はアルゴリズム設計者に委ねられており、自動化されていない。
## 関連
- 概念: [[Parameter Server]]
- エンティティ: [[Mu Li]] / [[Alexander J. Smola]] / [[David G. Andersen]] / [[Google]] / [[Carnegie Mellon University]] / [[Baidu]]
## 出典
- 本文: `.raw/papers/osdi14-parameter-server.txt`(全ページ精読)
- 図: `.raw/papers/osdi14-parameter-server/images/`(ページレンダリングからキャプション座標クロップ)
- URL: https://www.usenix.org/conference/osdi14/technical-sessions/presentation/li_mu