* この投稿は米国時間 5 月 18 日、Senior Software Engineer である Eugene Kirpichov と、Software Engineer である Malo Denielou によって投稿されたもの（投稿はこちら）の抄訳です。


はじめに

今回は、Google Cloud Dataflow において手動のチューニングを不要にする自動化機能について解説します。

以前、このブログ シリーズで Cloud Dataflow の自動スケーリング機能を紹介しました。この機能は、パイプラインのニーズに合わせてワーカー数を動的に調整します。今回の投稿では、動的なワーク リバランシング（Google 社内ではリキッド シャーディングと呼ばれています）を取り上げます。この機能はワーカーをビジーに保ちます。

この機能が、ストラグラー（担当する作業を完了するのに長い時間がかかるワーカー。ジョブの完了を遅らせ、他のリソースをアイドル状態にする）の問題にどのように対処し、多くのシナリオで大幅なパフォーマンス向上とコスト削減を実現するのか、また自動スケーリングとどのように連携するのかについて解説します。


ビッグデータ処理システムにおけるストラグラーの問題

すべての主要な分散データ処理エンジン（Google のオリジナルの MapReduce から、Hadoop、現代的なシステムである Spark、Flink、Cloud Dataflow まで）における重要な操作の 1 つとして Map があります。Map は、入力のすべての要素に関数を並列に適用します（Apache Beam Incubator プロジェクトのプログラミング モデル用語では ParDo と呼ばれます）。

これらのフレームワークはすべて、入力項目をそれぞれほぼ同サイズの複数の部分（多くの場合、シャードやパーテイションと呼ばれます）に分割し、各部分内のデータの読み込み / 処理を並列に行って Map ステップを実行します。

たとえば、glob ファイル パターンで指定された入力は < ファイル名, 開始オフセット, 終了オフセット > のタプルに分割され、Google Cloud Bigtable のようなキーバリュー ストレージ システムからの入力は < 開始キー, 終了キー > のタプルに分割される可能性があります。

Cloud Dataflow を除くそれらすべてのフレームワークの場合、分割はあらかじめシャードの実行前に行われ、実行中は分割状態は変わりません。シャードの数は通常、ユーザーによって指定されるか、システムによってヒューリスティックに決まります。たとえば、データ サイズの見積もりや、入力ファイルの数などに基づいて決定されます。

すべてのフレームワークが Map ステップで直面する非常に一般的なパフォーマンス問題が、ストラグラーの問題です。すなわち、少数のシャードの処理に対して、他のシャードよりもはるかに長い時間がかかってしまうという問題です。

ストラグラーは Map ステップのランタイムを長時間占有し、並列処理のメリットの多くを損なってしまいます。また、リソースを消耗し、コストを増大させます。他のワーカーはストラグラーのせいで、次のステージで作業を開始できるようになるまで長時間アイドル状態になってしまうからです。

ストラグラーは業界と学術界のいずれでも、低いデータ処理パフォーマンスの主因の 1 つと認識されています（ [1] [2] [3] ）。

ストラグラーの問題が発生しているジョブの例を 2 つ示します。ここでは、一般的な可視化手法の 1 つであるガント チャートで作業の実行状況を図示します。X 軸（横軸）は時間、Y 軸はワーカーです。緑のバーはそれぞれ入力の一部を処理するワーカーに対応し、空白部分はアイドル状態を示します。




不均等なデータ分割により、2 つのストラグラー タスクに長時間を要している単純なジョブ（読み取り、ParDo、書き込み）。ストラグラーを例示するために、動的なワーク リバランシングは無効化されている。



標準的な MapReduce ジョブを 400 ワーカーで実行。
（Google Cloud Storage から読み取り、GroupByKey を適用し、Cloud Storage に書き込み）
ストラグラーを例示するために、動的なワーク リバランシングは無効化されている。
X 軸 : 時間（最長 20 分）、Y 軸 : ワーカー


ストラグラーにはさまざまな原因があります。たとえば、特定のワーカーが異常に遅いことや、データの特定部分の計算負荷が特に高いこと、入力の不均等な分割によってさまざまな部分ごとにデータ量が著しく異なることが挙げられます。

このうち最後のケースは、ファイルを処理する際は滅多に発生しませんが、キーバリュー ストアなどの複雑な入力ではかなりの頻度で発生します。入力を均等に分割するためのデータ分布に関する十分な情報が前もって得られないからです。

別のタイプの問題として、データ量は均等に分割されているものの、処理の複雑さが一様ではないという場合があります。

たとえば、コード変換されるビデオのファイル名を含むファイルを入力として受け取るパイプラインがあるとしましょう。負荷が均等に分散されたワークフローを実現するためには、各ワーカーに同じ数のビデオを割り当てるのは適切ではありません。ビデオの長さが非常にまちまちである可能性があるからです。

ストラグラーが存在するということは、逐次実行されて長時間を要する作業が並列ジョブの中で事実上発生しているということです。アムダールの法則により、その影響はワーカーの数が増えるとともに、より顕著になります。


そのため、この問題の解決が Google の重要課題となったのは当然のことでした。Google では、人々が日常的に数千コアの規模でジョブを実行しているからです。Cloud Dataflow が誰でもオンデマンドでこの機能にアクセスできるようにしているので、ストラグラーの問題を適切に解決することは、優れたユーザー エクスペリエンスを提供するための必要条件でもありました。


ストラグラー対策の現状

業界と学術界は、ストラグラーの問題に長年取り組んできました。

  • 多くのフレームワークが、入力を分割する数を調整する仕組みを提供しています（たとえば、Cloudera のこの投稿や、Spark の正式なドキュメントは、Spark の並列処理のチューニングに関する幅広いアドバイスを提供しています）。一般に、細かく分割するとストラグラーは少なくなります。ただし、ある一線を超えると、分割数の多さによるオーバーヘッドの影響が出始めます。手動によるチューニングのデメリットは明らかです。手間がかかり、正確さに欠けます。さらに、入念に選んだ値がすぐに古くなってしまうおそれもあります。その原因としては、データセットに対する変更（特に、同じジョブがさまざまなデータセットに対して実行されている場合）、処理アルゴリズムに対する変更、ランタイム環境に対する変更、さらにはフレームワークの新バージョンへの移行が挙げられます。
  • 一部のフレームワークは低速なワーカーを特定し、そのマシンを不良と見なして、それらをワーカー プールから予防的に排除します。しかし、この解決策はハードウェア ベースのストラグラーにしか対応しておらず、クラウド環境にはあまり適しません。
  • 多くのシステムは投機的実行（バックアップとも呼ばれます）を実装しているか、または異常に遅いシャードを再起動します。これらの技術もあくまで低速なワーカーへの対策であり、不均等なデータ分割に対処するものではありません。
  • サンプリングや類似の技術は、データの分布を推測して、データをより均等に分割するのに役立ちます。しかし、そうした技術による統計データは多くの場合、収集コストが高くつくか、正確さが足りないか、あるいは古いものです。さらには前述したように、データが均等に分割されていても、それらのさまざまな部分における処理の複雑さが一様ではないために、ストラグラーが発生することもあります。
  • 学術界では、不均等なデータ分割に対処するため、より複雑な技術も研究されています。たとえば SkewTune は、ストラグラーの処理されていない入力を対象に統計データを収集し、この入力をより均等に分割します。これは正しい方向への一歩ですが、ストラグラーの未処理の入力を完全に読み取るには多大なコストがかかります。特に、ストラグラーが処理の一部を担当する入力の読み取りコストがもともと高い場合は、この技術は非現実的な選択肢になってしまいます。


動的なワーク リバランシングによるストラグラー対策

Google のサービスは他のデータ処理エンジンと同様に、さまざまなヒューリスティクス（発見的方法）を利用して、まず入力を複数の部分に分割します。しかし、さまざまなユースケースの大規模システム（たとえば Cloud Dataflow）を運用してきた経験から、私たちは次のような理解に達しています。
  • 前もってヒューリスティックなチューニングをどれだけ行っても（手動か自動かにかかわらず）、良好なパフォーマンスを保証するのに十分ではない。システムは実行時に、予測できない状況に必ず直面することが、その理由である。
  • 悪い状況に動的に適応し、そこから脱却できるシステムのほうが、ヒューリスティックな方法でそうした状況を回避しようとするシステムよりも、はるかに強力である。

私たちはこうした理解に基づき、Cloud Dataflow が実行中にワーク スティーリングやワーク シェディングと似た技術を使用して、分割を調整し続けるようにしています。このサービスは常に各ワーカーの処理の進捗状況を監視し、ストラグラーを特定します（予想完了時間が平均を上回っているワーカーなど）。

次に、Cloud Dataflow はこのようなワーカーに対して、未処理作業の一部（ファイルやキー レンジのサブレンジなど）を切り離すよう求めます。さらに、こうした新しいワーク アイテムをアイドル ワーカーにスケジューリングします。

そして、特定されたストラグラーを対象に、このプロセスが繰り返されます。ワーカーから切り離される作業量は、そのワーカーが残りの作業を十分に速く完了し、ストラグラーでなくなるように選択されます。


2 回にわたる分割の調整。すべてのシャードが現在の平均完了時間内に完了するように、ワーカーがアイドルになるとストラグラーのシャードが分割される。


この手法により、Cloud Dataflow は動的にストラグラーをなくし、作業を分割し直して、最適に近い並列処理とリソース利用を実現します。


技術的な課題

この手法は驚くほどシンプルに思えます。では、他のデータ処理エンジンでまだ実装されていないのはなぜでしょうか。結局のところ、この技術の堅牢な実装を実現するのは難しいからです。

第 1 の課題はデータの一貫性です。分割が前もって行われる場合、ワークフローがすべてのデータを 1 回だけ処理するようにするのは簡単です。

しかしこれは、データが同時に読み取られているときや処理されているときに分割を調整する場合は、はるかに難しくなります（ですが、私たちが発見したところでは、読み取りや処理に非常に長い時間がかかるレコードがある場合、良好なパフォーマンスを実現するには、そうした待ち時間のない分割が不可欠です）。


ワーカーやサービスの障害時も、ワークフローがすべてのデータを 1 回だけ処理するようにするのは困難です。データのすべての部分が追跡され、どのような状況でもレコードが 1 つも失われない（または重複しない）ようにする必要があります。パフォーマンスの観点からは、データの余分な読み取りや処理を回避することも望ましいです。

データの一貫性を確保するには、体系的かつ徹底的なテストや、多様な入力タイプすべてについて（オフセット ベースのテキスト ファイル、ブロック ベースの Avro ファイル、キーレンジ ベースの Bigtable など）、動的な分割が行われる場合の一貫性要件を一般的な形で示せる綿密な設計が必要になります。

これは厄介な課題です。Cloud Dataflow では、ソースは特定のものに限定されているわけではないからです。Cloud Dataflow（と Apache Beam）の各ソース API の実装は、すべてソースとなります。すべてのユーザーから「簡単に自分のソース用に正しく実装できる」と思っていただけるパブリック API を開発するのは困難です。


そこで私たちは、自動的で使いやすい包括的なテスト ツールを提供し、ユーザーを支援しています。これらのツールではソースは限定されておらず、数行のコードにより、API が正しく実装されているかどうかを徹底的にチェックできます。

もう 1 つの課題は、ストラグラーのタスクを特定し、分割の最適な調整を計算することです。これは根本的には、タスクが時間とともにどのように進行するかを予測するということです。

入力（ファイルなど）を一定のペースで処理するタスクであれば、それは簡単です。しかし、それがかなり困難になることもあります。


具体的には、ソースのデータ分布が不均一な場合（Google Cloud Bigtable など）や、レコード処理時間がワーカーによって著しく異なる場合、レコード処理時間が極端に長い場合など、タスクの進行に関する信頼できる情報が本質的に欠けている状況においてです。

こうした状況では、Cloud Dataflow はトレードオフの関係にあるスピードと分割精度のいずれかを優先する選択を動的に行う必要があります。

これらの課題に対処しなければならないことから、動的なワーク リバランシングは、Cloud Dataflow で最も複雑かつユニークな機能の 1 つとなっています。

このように、この機能は課題への対処が必要で、私たちにとっては厄介ですが、それでも私たちが提供しているのは、この機能の価値あるメリットをお客様に享受していただくためです。さらに、この機能における私たちの経験が他のシステムの教訓にもなれば、私たちにとってはうれしい限りです。


動的なワーク リバランシングのメリット

動的なワーク リバランシングには、ユーザーにとっての明確なメリットに加え、明確ではないものの非常に大きなメリットがあります。


  • ワーカーの稼働率の向上により、当然のことながら完了時間全体が短縮されます（同じ量の作業がより効率的に並列処理されるため）。
  • ワーカーが常にビジーとなるので、コストが大幅に削減されます（アイドル ワーカーはリソースを浪費します）。
  • 信頼性が高く、最適に近い並列処理が実現されるので、予測可能性も向上します。すなわち、ジョブの完了時間のばらつきが小さくなります。
  • 完了時間は、行われる計算量によって決まります。いくつかのワーカーに影響する可能性がある環境要因や、各ジョブの実行においてシステムのヒューリスティックな推測がどれだけ正確かということでは決まりません。
  • ユーザーは、並列処理やシャーディングのチューニングについて心配せずに済むようになり、コンピューティング アプリケーションのデータとビジネス ロジックに集中できます。
  • データセットの規模の大小にかかわらず、またデータ分布がまったく異なるデータセットでも、それらが同時に組み合わされている場合でも、あるいはユーザーが処理アルゴリズムを変更しても、同じパイプラインが最適な実行を継続し、チューニングやメンテナンスは不要です。
  • 最後に、動的なワーク リバランシングは自動スケーリングの実現に不可欠です。Cloud Dataflow が、たとえば 3 つのワーカーでパイプラインの実行を開始し、実行中に計算量の増大に伴って 30、120、500 ワーカーにスケールアップしなければならないと認識するのは、珍しいことではありません。


もちろん、500 ワーカーでの実行には入力のきめ細かな分割が必要になります。最新のリソース割り当てに応じて分割を調整することができなければ、ストラグラーのせいでスケーラビリティがすぐに損なわれてしまいます。

Cloud Dataflow はそうした調整を動的に行うことができ、3 ワーカー上ですでに実行された作業の一部が失われたり、複製されたりすることはありません。


パフォーマンスの向上例

説明はこれくらいにして、動的なワーク リバランシングによって実際に得られるパフォーマンスの向上を見ていきましょう。

この例では、データ分布が不均等な同じジョブを、動的なワーク リバランシングを有効または無効にして実行しています。動的なワーク リバランシングによってストラグラーが解消され、2 倍以上のスピードアップが得られることがわかります。
・データ分布が不均等なジョブを 24 ワーカーで実行（自動スケーリング : 無効、動的なワーク リバランシング : 無効）
・実行時間 : 最長 9 分

・同上（動的なワーク リバランシング : 有効）
・実行時間 : 最長 4 分 20 秒


次に示す例は、より大規模です。標準的な MapReduce 型のジョブを、n1-standard-4 マシン 100 台を使って実行します。Google Cloud Storage 上のファイルからデータを読み取り、キーによってグループ化し、処理を行い、結果を Google Cloud Storage に書き込んでいます。

このシンプルな構造から、最初に均等なデータ分割が行えます。それでも、このジョブにはストラグラーが発生しています（おそらく原因は、ワーカー間でのパフォーマンスの若干の違い、処理するデータの違い、Google Cloud Storage 自体のパフォーマンスのばらつきでしょう）。動的なワーク リバランシングで速度が最大 25 % 向上しています。

・MapReduce 型のジョブを 100 台の n1-standard-4 ワーカーで実行（動的なワーク リバランシング : 無効）
・実行時間 : 最長 20 分

・同上（動的なワーク リバランシング : 有効）
・実行時間 : 最長 15 分


この最後の例では、動的なワーク リバランシングが自動スケーリングと連携して機能しています。大規模なジョブの実行が 3 つのワーカーで開始され、サービスが計算量の増加を認識するのに応じて徐々に 1000 ワーカーまでスケールアップし、動的に分割を調整して最適な稼働を継続しています（青い縦線は、動的なリバランシングが相次いで作動し、タスクが分割されたことを示します）。





まとめ

今回は、Google Cloud Dataflow のユニークで重要な機能の 1 つである動的なワーク リバランシングを紹介しました。ストラグラーは、大規模なデータ処理システムで頻繁に発生する問題であり、その影響は数千コア（Cloud Dataflow では非常に容易にアクセスできます）にスケーリングしたときに特に大きくなります。

ストラグラーの原因がハードウェア異常、ネットワーク状況、不均等なデータ分布のいずれであるかにかかわらず、Cloud Dataflow の動的なワーク リバランシングは、ワーカー全体による最適に近い並列処理を維持します。

これによって、より高速な動作、より予測可能な実行時間、コストの低減、リソース使用率の向上、数千コアへの透過的なスケーラビリティが実現されます。しかも、こうしたメリットは、パイプラインをさまざまなデータセットに適用しても、ビジネス ロジックを変更しても、常に提供されます。

動的なワーク リバランシングは、私たち Google のユーザーが重要なもの、すなわちアルゴリズムとデータに集中できるようにする、もう 1 つの方法なのです。