BigQueryの分散処理の仕組みを深掘りする
Posted on
株式会社プレイドのCore Platformチームに所属しているBrownです。これは社内勉強会をブログ化したもので、BigQueryを題材にし、大規模なデータベースでの分散処理の仕組みについて紹介します。本ブログではBigQueryを題材にし、実際の大規模なデータベースでの分散処理を確認していくため、初学者の方から、もう一歩踏み込んでデータベースや分散処理の仕組みについて学習したい方を主な読者として想定しています。
弊社ではBigQueryについてのブログを複数執筆しており、このブログは2016年に現CPOである柴山が書いた「Bigqueryの内部処理について徹底解剖してみた」の2024年バージョンです。前回のブログでは内部でのクエリ処理に焦点を当てていますが、今回はBigQueryの全体のアーキテクチャを広く見ていきます。
BigQueryでの分散処理
BigQueryはリレーショナルデータ構造を扱うエンタープライズ向けのデータウェアハウスです。一般的なRDBMS（Relational DataBase Management System）と同様に、SQLを実行してデータベースに問い合わせを行いますが、そのアーキテクチャが一般的なRDBMSとは大きく異なり、ノードとストレージがそれぞれ大規模に分散して処理を行うことができます。ノードとストレージがそれぞれ独立してスケールできることで、負荷分散のパフォーマンス改善だけではなく、コストの最適化を行うこともできます。
BigQueryは以下のような複数の構成要素により成り立っており、ここでは、SQLのステートメントを細かい単位に分割して並列実行するDremel Execution Engine、高速なStorageとComputeを繋ぐネットワークであるSeparation of Storage & Compute (Jupiter network)、Storageレイヤでの処理を高速化するBigQuery Storage Engine(Colossus)の3つのコンポーネントについて、分散処理やキャッシュによるスループット向上の観点から深掘りして考察を進めます。
[1]
Dremel Execution Engine
Dremelは「Dremel: Interactive Analysis of Web-Scale Datasets(2010)」と「Dremel: a decade of interactive SQL analysis at web scale(2020)」の論文の中で詳細にそのアーキテクチャが記載されており、この章ではそれらの論文をもとに考察を進めます。以下の説明はDremel: Interactive Analysis of Web-Scale Datasets(2010)の6. QUERY EXECUTIONの章と、Dremel: a decade of interactive SQL analysis at web scale(2020)の5.2 Evolution of serverless architectureの章を読んで自分なりの理解を書いてみました。間違っていたらすいません。
まず、2010年に発表された初のDremel論文では以下のようにクエリを解釈して分散処理をする流れが記述されています。
[2]
次のSQL
SELECT A, COUNT(B) FROM T GROUP BY A
をBigQueryに対して、実行すると、まずそのSQLをroot serverが受け取り、テーブルTを構成する水平に分割されたテーブルの全てを決定して、次のようにクエリを書き換えます。
SELECT A, SUM(c) FROM (R_1 UNION ALL ... R_n ) GROUP BY A
（⚠︎現在使っているブログツールの仕様上の問題で数式の記述が難しかったので、元の論文から記述を変更しています）
ここで、テーブルR_1 …R_nはノード1…nに分かれて送られたクエリの結果を意図し、T_i をTの分割されたテーブルのi番目の要素とすると、R_iは以下のように表すことができます。
R_i= SELECT A, COUNT(B) AS c FROM T_i GROUP BY A
DremelはQuery dispatcherでR_iのクエリの優先度や負荷のバランスからスケジューリングをして、それぞれ分割してLeaf serverで実行します。Leaf serverではさらに複数のスレッドに分かれてプロセスが実行されており、Leaf serverとそれぞれのLeaf serverのスレッド数をかけたものの総和をSlot数と呼びます。Slot数はLeaf serverの数×Leaf serverのスレッド数で表すことができるため、例えば、3,000個のLeaf serverがあり、それぞれが8スレッドで処理していた場合、スロット数は24,000個となります。
次に、2020年に発表された改善されたDremelについて言及されている、最新のDremel論文の内容を見ていきます。2010年のDremel論文で発表された内容からクエリ処理エンジンの部分で改善が入っています。
次のクエリを実行すると、以下のようなステップでクエリが処理されます。
SELECT language, MAX(views) as views
FROM wikipedia_benchmark.Wiki1B
WHERE title LIKE "G%o%"
GROUP BY language ORDER BY views DESC LIMIT 100
- クエリを実行すると、まずCordinatorがそのクエリを受けとり、実行計画を構成するCoordinatorはExecution Treeの形でDAGを構成する。WorkerはPool内に待機しており、DAGが形成されると、各Workerに具体的なタスクが割り当てられる
- DAGを元に、Stage 1のLeafでFilter, 部分的な GROUP BYが実行され、Shufflingを経て、異なるサーバーのインメモリに配置される。これにより、データはHash partitioningを通じて適切に分散され、Stage 2でPartation keyごとに最終的なGROUP BYや、LIMIT、SORTなどが適用される
- Stage 3では最終的なSORTやLIMITなどを適用し、その結果をShufling persistant layerへ書き込みを行う
- CoordinatorはそのShufling persistant layerへ書き込まれたデータをLIMIT 100で100件読み取りリクエスト元のClientへ返す
上記の説明を図にしたものは以下で、Carnegie Mellon大学の講義資料として公開されているものがわかりやすかったので引用します。
[4]
2010年時のDremelとの比較すると、実行中にデータのカーディナリティの誤推定があった場合に、実行計画が動的に変更されるようになっています。これにより、実行計画を動的に最適化できるため、より高速なクエリ実行が期待できます。
BigQuery Storage Engine(Colossus)
この章では、BigQueryのデータ実体の入るストレージレイヤの考察を「CacheSack: Theory and Experience of Google’s Admission Optimization for Datacenter Flash Caches(2023)[5]」の論文を元に行います。
Colossusのアーキテクチャは上記の図のようになっており、このアーキテクチャについて解釈と考察を進めます。
Dremelが分割されたクエリR _nを処理する過程でストレージへのアクセスが生じます。この時に、分割して複数のSlotから実行されるクエリによりストレージに都度アクセスを行っていると、ストレージI/Oによるレイテンシー増加や、ストレージの高負荷によるパフォーマンス劣化が発生します。そこで、このColossusのアーキテクチャでは以下のようなリクエストの処理を行うことで、キャッシュ戦略と負荷分散を行っているようです。
- Colossus File Cache ClientがColossusにあるデータへアクセスを行うときに、まずはCache Index Serverへ、そのデータのキャッシュがあるかRPCを送る。もしキャッシュがある場合にはキャッシュの所在地のIndex情報を返し、もしキャッシュがない場合には、キャッシュがないことをColossus File Cache Clientへ返す
- Colossus File Cache Clientは1でキャッシュのIndexの情報を得た場合にはその情報を元に、Flash Serverにあるキャッシュされたデータを取得し、もし1でキャッシュがなかった場合には、Disk Serverからデータを取得する
- 2でDisk Serverからデータを取得した時に、Cache Index Serverがその情報をキャッシュしておくべきだと判断した場合にはFlash Serverへそのデータをキャッシュし、Cache Index ServerへそのキャッシュのIndex情報を保存する
Flash Cacheに作られたデータのキャッシュはいつまでも保持しておくわけではなく、LRUで削除されていきますが、LRUで都度キャッシュを消していくと不規則にFlash Serverへ書き込み処理が走り、SSDへの書き込み増幅[7]が起こることでパフォーマンスが劣化するため、Flash ServerでFIFOの複数の通常1GiBの固定ファイルサイズのColossus Fileで構成される。
キャッシュのキューを持ち、そのキューをFIFOでキューイングされたものを追い出すことで、書き込み処理の頻度をコントロールしながらキャッシュを消しているようです。
ここで私が気になった点としては、ディスクサーバーではHDDを使い、Flash ServerではSSDを使っている点です。Flash ServerにキャッシュしたデータはSSDを使うことで、コストはHDDに比べて高いものの、シークタイムなどを削減できるため、パフォーマンスは期待できます。しかし、全てのデータをSSDに保存するとコストが高くなるため、キャッシュに乗っておらず、頻繁にアクセスされるデータではないものはHDDに保存することで、コストのバランスを取ることを目的としていると考えました。
随分とSSDの値段が下がってきましたが、今後さらに大容量で安価なSSDが主流になってくると、Disk Serverの部分もSSDに変わり、よりBigQueryのクエリが高速化されるのでしょうか。
Compute (Jupiter network)
最後に、DremelとColossusを繋ぐJupiter networkについて、「A look inside Google’s Data Center Networks(2015)」を元に考察を進めます。
先に見たように、DremelとColossusは分離したコンポーネントにより構成されており、それぞれのコンポーネントの中で速く処理が行われていても、そのコンポーネントをつなぐネットワークが遅ければ、処理全体が遅くなってしまいます。そこで、BigQueryではJupitor networkと呼ばれる非常に高速なネットワークシステムが使われており、上記の投稿にはそのスループットがこのように書かれています。
Our current generation — Jupiter fabrics — can deliver more than 1 Petabit/sec of total bisection bandwidth. To put this in perspective, such capacity would be enough for 100,000 servers to exchange information at 10Gb/s each, enough to read the entire scanned contents of the Library of Congress in less than 1/10th of a second
現在の世代である Jupiter fabricsは、1Petabit/秒以上の総二分バンド幅を提供できる。これは、10万台のサーバーがそれぞれ10Gb/秒で情報交換するのに十分な容量であり、国会図書館のスキャンされたコンテンツ全体を1/10秒未満で読み取るのに十分な容量である
このように、BigQueryの内部でのネットワークも非常に高速であることから、複数のコンポーネントに分かれていて、さらにそのコンポーネント間でネットワーク通信が発生する場合でも、高速なクエリ処理が実現されているものと考えています。
終わりに
ここまでBigQueryに関連した複数の記事や論文から、特に分散処理に関連する項目についてピックアップして考察を進めました。プレイドでは、GCPの東京リージョンができるよりも前からGCPのリソースを多用しており、BigQueryやBigTableへも日々大量のデータを保存して、活用しています。プレイドのKARTEで計測するエンドユーザーのイベントが保存されているBigQueryのテーブルへの書き込みだけでも10TB/日を超えており、全体ではPB級のデータがBigQueryに蓄積されています。大規模なデータではあるものの、それをただ蓄積していくだけでは意味がなく、それを価値のある、利用できるデータへ変えていくことが不可欠です。 我々は、その過程でより速いクエリや、システムのアーキテクチャを考えていくにあたり、よりBigQueryの内部の設計の理解を深め、その特性をうまく活かせるような設計や実装を行なっています。
マネージドサービス使うことが主流になり、コンソール画面からボタンをクリックしていくだけでリソースが立ち上がり、そのシステムの中身を知らなくても、ドキュメント通りに操作すれば利用可能というクラウドサービスが増えてきました。しかし、私はそのコンピューティングリソースの持つ性能を最大限発揮したり、ディザスタリカバリの速度や安定性も考慮した堅固なシステムを設計する際には、そのクラウドのコンピューティングリソースの内部についてディープダイブして理解した上で、自身のサービスの設計をすることが必要だと考えています。
GCPでは今回紹介したいくつかの論文のように、それぞれのサービスの詳細な設計や特性が理解できる論文がUSENIXなどの学会で発表されるため、それを元に、ディープダイブして理解できるようになっています。システムを作った上で、それを論文にして発表していくことはコストのかかる活動だと思いますが、引き続き継続していただけると、ユーザーとしては非常に嬉しいです。
参考・関連ドキュメント
OLAPデータベースにおける高速化の技術
Capacitorの基礎技術であるColumnar data formatなどについてはこちらもご覧ください
https://tech.plaid.co.jp/fundamentals_of_olap_db_technology
Dremel: Interactive Analysis of Web-Scale Datasets
Dremel: a decade of interactive SQL analysis at web scale
Carnegie Mellon University, Lecture #17: System Analysis (Google Dremel / BigQuery)
Google Dremel - a Decade Later
CacheSack: Admission Optimization for Google Datacenter Flash Caches
The 12 Components of Google BigQuery
A look inside Google’s Data Center Networks
Inside Capacitor, BigQuery’s next-generation columnar storage format
Sergey Melnik, Andrey Gubarev, Jing Jing Long, Geoffrey Romer, Shiva Shivakumar, Matt Tolton, Theo Vassilakis Proc. of the 36th Int'l Conf on Very Large Data Bases (2010), pp. 330-339 https://research.google/pubs/dremel-interactive-analysis-of-web-scale-datasets-2/ ↩︎
Gubarev, Andrey, et al. "Dremel: A Decade of Interactive SQL Analysis at Web Scale." (2020). ↩︎
Andy Pavlo,CMU 15-721,Spring 2024; Google Bigquery / Dremel https://15721.courses.cs.cmu.edu/spring2024/slides/17-bigquery.pdf ↩︎
Tzu-Wei Yang, Seth Pollen, Mustafa Uysal, Arif Merchant, Homer Wolfmeister, and Junaid Khalid. 2023. CacheSack: Theory and Experience of Google’s Admission Optimization for Datacenter Flash Caches. ACM Trans. Storage 19, 2, Article 13 (May 2023), 24 pages. https://doi.org/10.1145/3582014 ↩︎
Tzu-Wei Yang, Seth Pollen, Mustafa Uysal, Arif Merchant, Homer Wolfmeister, and Junaid Khalid. 2023. CacheSack: Theory and Experience of Google’s Admission Optimization for Datacenter Flash Caches. ACM Trans. Storage 19, 2, Article 13 (May 2023), 24 pages. より引用 ↩︎
Write amplification, Wikipedia https://en.wikipedia.org/wiki/Write_amplification ↩︎
https://cloudplatform.googleblog.com/2015/06/A-Look-Inside-Googles-Data-Center-Networks.html ↩︎