BigQuery のアーキテクチャの変遷を論文 Dremel: A Decade of Interactive SQL Analysis at Web Scale から読み解いてみた
はじめに
皆様、こんにちは。Google Cloud Japan Customer Engineer Advent Calendar 2020 の 8 日目は 今年公開された BigQuery のリサーチペーパーを読んでみて個人的に興味があった点をまとめてみようと思います。2020 年で 10 周年を迎えた BigQuery の進化の過程が理解できるので皆様もお時間あればご一読を！
TL;DR
このホワイトペーパーは、Dremel （BigQuery のクエリエンジン） が採用している主要なアーキテクチャや考え方（これらのいくつかはクラウドネイティブなデータウェアハウスではトレンドとなりつつあります）がこの10 年間でどのような進化を経て現在の BigQuery になったのかを、Seattle Report on Database Research というレポート内で述べられている主要な 5 つのテクノロジーのトレンドの観点からまとめた内容となっています。
- SQL : 拡張性のために SQL から一時期離れていた Google がどのようにして BigQuery で再導入して進化させていったか。SQL の移植性に対する考え方も垣間見れて興味深いです。
- Disaggregated compute and Storage（コンピュートとストレージの分離）: 今や業界標準となりつつあるコンピュートとストレージの分離だけでなく、メモリの分離に至るまでの改善、課題、また改善のサイクルが見えてこれまた面白いです。
- In situ analysis: 訳が難しいですが、これは Data Lake、BigQuery からデータを移動させずに相互にアクセス、分析可能な仕組みのことです。Big Data 分析ではコスト、データ移動の時間を考えても非常に重要な観点と言えます。
- Serverless computing（サーバーレス コンピューティング）: Google 内の数千人のユーザに低コストで拡張し、高速なクエリ分析を提供するために既存の DWH とは異なるアプローチを可能にしたコアのアイデアや進化の過程がわかります。様々な DWH を触ってきたという方にこそ読んでほしい章です。
- Columnar storage（列指向ストレージ）: 列指向ストレージは多くのデータウェアハウスで利用されていますが形式は複数あります。これらの種類と BigQuery で採用されている形式が効率性の面でなぜ優れているのかなどの詳細について述べられています。
これ以外にも Interactive Query Latency Over Big Data というビッグデータに対するインタラクティブなクエリを高速化する技術についての章があり、大規模なデータセットに対するクエリがBigQuery でなぜ高速に動作するのかが理解できる内容となっています。
実際の内容に入る前におさらいとしてストレージ、コンピュート、メモリが分離された現在の BigQuery のアーキテクチャもこちら（図 1）に掲載しておきます。
では、このブログでは Disaggregated compute and Storage と Serverless computing の2つの章について詳細をみていきたいと思います。
Disaggregation （分離）
この章はストレージの分離、メモリの分離と分離によるメリットの３つで構成されています。
Disaggregated Storage （ストレージの分離）
まず最初に、2006 年に 20% プロジェクトとして Dremel が考案された時は一般的な DWH と同じくシェアードナッシングアーキテクチャを採用し、各サーバにあるローカルディスクに保存していました。分析システムでパフォーマンスをだすにはこの構成がベストプラクティスであった時代になります。しかし、この構成では、Dremel のワークロードが増えるにつれて管理が困難になり新たなアーキテクチャに移行していきます。
2009 年に Borg と呼ばれるクラスター管理システムに移行されました。Borg といえば Kubernetes の元となったコンテナオーケストレーションの仕組みです。ワークロードの増加のため Borg に移行したわけですが、共有リソースであるがゆえに他のジョブとのスピンドルの共有という新たな課題が発生しました。
課題についての詳細は書かれていませんが、次の対応の効果として速度とスケーラビリティの向上とあるので性能的な課題と考えられます。
ジョブ間のスピンドル共有という課題に対し、分散ストレージの仕組みを変更することで対応しました。具体的には、独立したサーバで管理され、テーブルを３つのローカルディスクに保存する、複製されたストレージ組織に移行しました。これによってペタバイト規模、数兆行レベルのテーブルを扱えるようになりました。しかしながら、ローカルディスクへの複製の保存がストレージと処理を密結合することになり、それによってすべてのアルゴリズムが複製対応する必要があり新しい機能の追加が困難になり、なおかつストレージの拡張にサーバの追加が必要になりデータがロックアップされ、Dremel 以外の方法でアクセスできなくなりました。
この課題に対して、Google のストレージとネットワーク ファブリックが劇的に改善されていたので再度シェアードナッシングアーキテクチャに移行します。最初の Google File System(以下、GFS)ベースの Dremel では、ファイルオープンの時間や Dremel のメタデータの形式がディスクベースでネットワークベースではないことに起因して大幅なパフォーマンスの低下が見られましたが、最終的には、ローカルディスクベースのシステムを上回るレベルにチューニングされました。
このように分散ストレージを利用したアーキテクチャへの移行により、データの分離と複雑さを減らすだけでなく、フルマネージドの GFS が Dremel の堅牢性と SLO を向上させ、GFS から Dremel のローカルディスクへの初期ロードを不要にし、Dremel を 他のチームが利用する際に利用開始が楽になるなどの効果もありました。後に GFS から Colossus に移行することでスケーラビリティと堅牢性がさらに向上しました。
Disaggregated Memory（メモリの分離）
BigQuery が結合などの Shuffle 処理に分散インメモリを利用していることは有名ですが、どのような過程を経て現在の仕組みになったのかを見ていきましょう。
初期の Dremel のシャッフルは、ローカルメモリとディスクに中間結果を保存していましたが、コンピュートノードとの緊密な結合がスケーラビリティのボトルネックとなりました。具体的には、データを前のステップから受け取り、処理をするコンシューマーと、処理したデータを次のステップのノードに渡すプロデューサーの増加によるシャッフル操作の二次関数的な増加を効率的に解消することができない、リソースの断片化と詰まりなどによるボトルネックが発生しました。
これに対して 2012 年に Colossus 分散ファイルシステムを使用して新しい分離されたシャッフルのインフラを構築したところ、Hyper Dimension Shuffle: Efficient Data Repartition at Petabyte Scale in SCOPE. と Riffle: Optimized Shuffle Service for Large-Scale Data Analytics. にかかれているすべての課題に直面しました。
この課題を解決するため、2014 年にインメモリのクエリ実行をサポートするシャッフルのインフラに落ち着きました。この実装では図 2（分散インメモリシャッフルの仕組み）のように中間のシャッフルデータを格納する RAM とディスクは分散型一時ストレージシステムで管理されています。
この分散インメモリによりシャッフルの待ち時間の 1 桁レベルの短縮、1桁 大きいサイズのシャッフル、リソースコストの 20% 以上の削減という効果をもたらしました。
Observations （分離することのメリット）
正確にいうと Observations は観測、監視になりますが、ここではリソースを分離することによる様々なメリットについて述べられているのでそのように訳しました。
分離することによってリソースのタイプごとに独立してプロビジョニングすることができ、コストとパフォーマンス と弾性が向上することや、規模の経済、普遍性（多くの分析システムに採用）、高レベルの API （分離されたリソースを抽象化することで暗号化、フィルタリング集計などの機能を組み込むことができる）、付加価値のある再パッケージ化という側面があります。
Disaggregation の章はいかがでしたでしょうか？ 最初は一般的なシェアードナッシングアーキテクチャを採用していたのは驚きでした。そこからマルチテナント、複製されたストレージ組織、GFS の分散ストレージ、最終的には Colossus と管理、性能、拡張性等様々な課題を解決しながら、10年以上かけてしっかりと洗練されて今日のアーキテクチャになってきたということがわかる内容となっています。分離することによる付加価値で述べられている高レベルの API の 1 つとして、Storage API というクエリエンジンを介さず BigQuery のストレージに直接アクセス可能でかつストレージ側でフィルタリング等のオフロードができるという機能もそのおかげなのかと感じるところもありました。
では続いては Serverless Computing についてです。
Serverless Computing （サーバーレス コンピューティング）
この章では現在では広く知られるようになったサーバーレスのルーツとアーキテクチャの進化の 2 つの構成となっています。
Serverless roots（サーバーレスのルーツ）
Dremel が作られた当時はデータウェアハウスは専用サーバを使っており、Hadoop など柔軟性のあるビッグデータのフレームワークでもシングルテナントの構成でした。そんな中で Google 内部の数千人のユーザに低コストで、インタラクティブで低レイテンシのクエリとその場での分析をサポートするにはマルチテナントでオンデマンドでリソースをプロビジョニングできることが必要でした。
サーバーレス分析を可能にするための 3 つのコアなアイデア
- 分離： コンピュート、ストレージ、メモリを分離することで、ストレージと独立してオンデマンドでの拡張とコンピュートの共有ができるようになり、より低いコストでユースケースに適応することが可能。
- フォールト トレランス（障害許容設計）と再起動性： コンピュートリソースが遅かったり利用できない可能性があり、ワーカーの信頼性が低いという仮定をもとに Dremel は構築されました。この仮定がクエリのランタイムとディスパッチのロジックにも大き影響を与えていました。
クエリの各サブタスクは失敗しても少量のタスクが別のワーカーで再開できるように決定論的で反復可能である必要がありました。
タスクのディスパッチャは応答しないワーカーの影響を軽減するために同一タスクの複数のコピーのディスパッチができる必要がありました。この仕組でサブタスクをキャンセルして再スケジュールすることができ、クエリに割り当てられたリソースの量を簡単に調整可能になりました。 - 仮想スケジューリングユニット： 特定のマシンタイプに依存するのではなく Dremel のスケジュール ロジックは、コンピュートとメモリを抽象化した単位の slot で動作するように設計されました。これによってスケジューリングと利用者にみえるリソースの割当をコンテナやマシンの形状およびサービスの展開から切り離すことができました。
残りの部分は、この3つのアイデアが多くのサーバーレス分析のビルディングブロックになっていることや、分離については産業界、学界で幅広く採用されていることなど Dremel のアーキテクチャが与えた影響について述べられています。
Evolution of serverless architecture（サーバーレス アーキテクチャの進化）
Dremel のサーバーレスの機能は継続的に進化し続け、BigQuery の重要な特徴の 1 つになっています。Dremel のアプローチを新しいアイデアに進化させ、サーバーレスを次のレベルにおしあげた新しいアイデアの紹介になります。
Centralized Schedulin（集中型スケジューリング）: 2012 年にこのスケジューリングに切り替わりました。これによってよりきめ細かいスケジューリングが可能になり特定の顧客にリソースを割り当てる Reservation を可能にさせました。図 3（クエリ実行のシステムアーキテクチャ）にある中間サーバのディスパッチャに代わり、クラスタ全体の状態をもとにスケジューリングが可能になり使用率と分離を向上させました。
Shuffle Persistence Layer （シャッフルの永続化層）: 2010 年の論文公開後に実装されたシャッフルは進化をして、クエリ実行における各ステージのスケジューリングと実行の分離が可能になりました。シャッフルの結果をクエリ実行のチェックポイントして使用することで、スケジューラが動的にワーカーをプリエンプションでき、コンピュートリソースが制約されている時に他のワークロードのためにリソース割当を削減することができるようになりました。
Flexible Execution DAGs （DAG の柔軟な実行）： 実行ツリーの固定は集計には向いていますが複雑なクエリーには理想的ではありません。集中型スケジューリングとシャッフルの永続化層に移行することで次のようにアーキテクチャが変化しました。
- クエリコーディネーターがクエリを受信し、実行ツリーの DAG であるクエリプランを構築しスケジューラから指定されたワーカーを使用しクエリの実行を指揮します。
- 構造が未定義な状態でワーカーはプールとして割当てられ、コーディネータが実行 DAG を決定するとすぐに実行可能なローカル実行プランをワーカーに送信します。リーフのステージのワーカーはストレージからデータを読みシャッフル永続化層に書き込み、他のステージのワーカーは永続化層を読み書きします。最終的に実行が終わると永続化層に保存されコーディネータがクライアントに通知します。
ウィキペディアのテーブルに対する top-k クエリの実行は次のように実行されます。（図4 : クエリの実行プラン例）
- ステージ1でワーカーは分散ストレージからデータを読み、フィルタを適用後、ローカルで部分的に集約して language 列をハッシュパーティションしてシャッフルします。
- ステージ2で最終的な集計を実行してから、別のキーでソートして、LIMIT で切り捨てた結果を次のステージに送信します。
- 1つのワーカーがシャッフル永続層から最終的なソートと切り捨てを行い結果をシャッフル層に書き込みます。
このようなクエリが1つから数万までの任意の数のワーカーで実行可能な柔軟性がシャッフル永続層によって提供されます。
Dynamic Query Execution （動的なクエリ実行）： データの形に応じてクエリエンジンが適用可能な複数の最適化の方法があります。例えば、Broadcast 結合と Hash 結合の選択を検討する場合、Broadcast 結合は、結合のプローブ側でデータをシャッフルする必要がないので非常に高速ですが、ビルド側がメモリに収まるほど小さい場合のみ利用できます。
クエリのプラン中に正確なカーディナリティを見積もるのは難しく、結合を通して誤差が指数関数的に増えることはよく知られています。Dremel はクエリの実行中に収集した統計をもとに動的にクエリの実行プランを変更することができます。これはシャッフルの永続化層とクエリコーディネータによる集中的なクエリ オーケストレーションによって可能になりました。Broadcast 結合と Hash 結合も両側でデータをシャッフルし Hash 結合で開始しますが、片側で早く処理が終了し、Broadcast のデータサイズのしきい値を下回ると 2 番めのシャッフルをキャンセルし Broadcast 結合を選択します。
BigQuery の大きな特徴であるサーバーレスのルーツと進化を論文を通じて簡単に紹介してみましたが、いかがでしたでしょうか？ ワーカーが失敗しても別のワーカーで処理が再開可能、シャッフルの永続層をチェックポイントとして動的にワーカーをプリエンプションできる機構などは高可用性やワークロード管理の面で大きな利点があると思います。またDAGの柔軟な実行、プランの動的変更は静的な統計からでは生成できないより良いプランで実行できるのでリソースの無駄な使用を抑え、クエリの性能向上のメリットがあると考えられます。
最後に
今回は Disaggregation、Serverless Computing の2つの章を取り上げました。論文で言及されている SQL、Columnar Storage、クエリのレイテンシー削減の仕組みの章も、技術的に面白い内容ですので興味を持たれた方は冬休みにでも是非、読んでいただければ幸いです。
Disclaimer : 本内容は個人の意見であり、所属する組織を代表するものではありません
