BigQuery によるインメモリ クエリの実行

* この投稿は米国時間 8 月 2 日、BigQuery Technical Lead である Hossein Ahmadi によって投稿されたもの（投稿はこちら）の抄訳です。


Google BigQuery は、高速を誇るペタバイト規模のアナリティクス データベースです。この規模に見合ったパフォーマンスを発揮できるように、クエリを完全にインメモリで処理します。

ほとんどのデータベースとデータ処理システムは、ディスクとメモリの両方を使うハイブリッド エンジンを使ってスケーラビリティを確保します。それに対し、BigQuery の実行エンジンは、シンプルで純粋なインメモリのオペレータを構築し、スケーラブルなデータの再分割、いわゆる “シャッフル” を通じてペタバイト規模のアナリティクスを実現しています。

本稿では、この BigQuery のシャッフルにスポットライトを当てます。Jupiter のようなペタバイト規模のネットワーク システムを活用し、ハイパフォーマンスなインメモリ クエリ実行を実現するシャッフルの仕組みについて説明します。

シャッフルは、Hadoop や Spark から Google Cloud Dataflow に至るまで、あらゆる分散データ処理システムに欠かせない重要な要素です。大規模で複雑な結合、集計、アナリティクス処理では、必ずシャッフル ステップが必要になります。

たとえば MapReduce は、マップの処理結果のキーに基づき、“マップ” から “リデュース” にデータを再分配するためのメカニズムとしてシャッフル ステップを使っています。大規模な計算をサポートするために、シャッフルされたデータは一時的にリモート サーバーに格納されます。マップとリデュースの分割は、これによって実現されているのです。

シャッフルは、大規模なデータ処理をサポートするための基本原則です。ただし、シャッフル ステップは遅いことで有名です。Hadoop のエンジニアたちは MapReduce のことを、冗談交じりに「本当なら小さなマップ、巨大なシャッフル、小さなリデュース処理と呼ぶべきだ」と語っています。

BigQuery にはさまざまな特徴と要件を持つクエリがあるため、シャッフル ステップの見直しが必要になりました。そうしたことから、BigQuery のシャッフルは、2014 年にメモリ ベースで（ディスク スプーリング サポート付き）、Google データセンターのネットワーク テクノロジーに特化して設計された新開発のインフラストラクチャに移行しました。

このインフラストラクチャは、ハッシュの結合など特定の分散処理だけでなく、さまざまなユースケースに対応する柔軟なデータ デリバリ システムとして設計されています。このプロジェクトは、何年にもわたるデータ転送テクノロジーの研究開発の集大成です。

BigQuery のインメモリ シャッフルは、リモート メモリのホスティング専用に作られた一連のノードに対し、クエリ処理のさまざまなステージで作られた中間データを格納します。Spark（RDD という形）や Piccolo（分散ハッシュ セットという形）など多くのシステムでは、一般にデータ処理ジョブの中間結果を永続記憶に書き込んでいますが、BigQuery はシャッフル処理と密接に統合された形で中間結果をインメモリに格納しており、他とは異なる方向を選んでいます。

BigQuery のシャッフルが定義するデータ再分割抽象は、再分割されるデータを作るプロデューサ（producer）、再分割されたデータを受け取るコンシューマ（consumer）、シャッフルを管理するコントローラ（controller）の 3 つのエンティティから構成されています。

Producer (producer_id) {  
  void SendRow(row, consumer_id) : Called to send a row to a given consumer  
                                   on behalf of this producer.
}
Consumer (consumer_id) {  
  string ReceiveRow() : Called to receive one row for this consumer.
}
Controller {  
  StartShuffle() : Called before any producers or consumers start sending or 
                   receiving rows.   
  EndShuffle()   : Called after all producers and consumers have successfully                   
                   sent and received all rows. 
}

この API は、適切なレベルで共有メモリ抽象を提供するように配慮して設計されています。あらゆる形態のデータ再分割のサポートと、データ処理パイプラインを使った転送をこなせる程度には汎用的ですが、Google のネットワーク テクノロジーを使用した効率的な実装のために特化しています。


シャッフル処理を図示すると、概念的には次のようになります。

上記の抽象によってサポートされている特定の処理セットのもとで、BigQuery のシャッフルは、多くのプロデューサが最小限のオーバーヘッドとフラグメンテーションでリモート メモリに書き込めるようにするとともに、コンシューマが高いスループットで同時に読み出せるようにしています。

具体的には、プロデューサは連続したメモリ ブロックで作成された行のログとインデックスを作ります。このインデックスにより、コンシューマは目的の行に効率よくアクセスし、読み出すことができます。また、BigQuery のデータ構造と通信プロトコルは、Jupiter ネットワーク インフラストラクチャが実現しているハードウェア アクセラレーションを利用しています。

BigQuery シャッフルが MapReduce スタイルのシャッフルと異なる点は、インメモリでのデータの再分割以外にもあります。それは、データフロー グラフのさまざまなステージでシャッフルが障害にならないことです。

MapReduce スタイルのシャッフルの場合、すべての行を送ってソートした後でないと、再分割されたデータセットの最初の行を使えません。これに対して BigQuery のシャッフルでは、プロデューサが作った出力はただちにシャッフルされ、BigQuery ワーカによって消費できる状態になります。これにより、パイプラインで分散処理を実行できるわけです。

データ分割は、BigQuery のクエリのパフォーマンスに非常に大きな影響を与えます。特に、最良の結果を得るためには、適切な数のコンシューマを使い、プロデューサからコンシューマにデータを渡すマッピング関数として適切なものを使うことが重要です。分割が最適でなければ、処理が何桁分も遅くなるばかりでなく、リソースの制約のために失敗する危険さえあります。

BigQuery は、クエリで使われる処理タイプ、データの量、バックグラウンドの負荷、その他の要素に基づいて、インテリジェントに分割方法を選択する動的なパーティション メカニズムを使っています。そのため、BigQuery ユーザーは、分配やソート キーを指定した際のオーバーヘッドをかけずに、あらゆるデータセットに効率よくクエリを実行できます。

BigQuery を使用すればペタバイトものデータにクエリを発行できますが、再分割されるデータがこれだけ多いと、多くの場合、純粋なインメモリ シャッフルはコスト的に実現不可能です。

BigQuery シャッフルは、リモート メモリから Google の分散ファイル システム Colossus に一時データを移動し、再構成するという方法でこの問題を解決しています。ディスクとメモリのパフォーマンス属性は根本的に異なるため、BigQuery は特別な配慮のもとにディスク シークを最小限に抑えるようにデータを自動的に構成しています。

BigQuery は、サイズの調整や仮想マシン クラスタのデプロイ、リソースのプロビジョニングを必要とせず、誰でもクエリを実行できるように作られているマルチテナント サービスです。

BigQuery シャッフルは、これを実現するために、ほとんどのクエリを純粋にメモリ内で実行できるようにするインテリジェントなメモリ リソース管理システムを使っています。このシステムは、クエリの負荷の違いに瞬時に適応します。

BigQuery のすべてのクエリは、もれなくシャッフル処理を必要としますが、1 行のデータを送るときも、あるいはペタバイトのデータにクエリを実行するときも、同じインフラストラクチャが使用されます。

BigQuery ユーザーは、極端なまでの柔軟性を持ち、Google のネットワーク テクノロジーに密接に統合された BigQuery シャッフルにより、あらゆる規模で高速なデータ アナリティクスを利用できるのです。


- Posted by Hossein Ahmadi, BigQuery Technical Lead