BigQuery Storage APIの速度比較をやってみる

ML関係のオペレーションや分析をやっていると、BigQuery上のデータを直接ローカルにダウンロードして処理することが多々あります。
最近は専らcolaboratoryを使って分析・可視化をすることが多いので、クイックに使えるマジックコマンドでデータをダウンロードすることが多いのですが、対象のデータが大きくなるとダウンロードに要する時間がかなり大きくなってしまいます。

そこで、Google Cloud Next '19で発表された新機能を紹介します！　(Cloud Run, BigQuery Storage API, Cloud Data Fusion)で紹介されていたBigQuery Storage APIでどれくらい時短になるのかを試してみます。

BigQuery Storage API

詳細は以下の記事を参考にしてください。

• BigQuery Storage API Overview
• BigQuery Storage API を使用して BigQuery データを pandas にダウンロードする
• Python Client for BigQuery Storage API (Beta)

インストール

BigQuery Storage APIを利用するには、BigQueryクライアントライブラリ バージョン 1.9.0 以降と BigQuery Storage API クライアントライブラリをインストールします。

pip install --upgrade google-cloud-bigquery[bqstorage,pandas]

速度比較

速度比較では、クエリからBQのデータを取得してpandas.DataFrameとしてダウンロードするまでを比較します。

データ分析タスクでは、BQテーブルからクエリを使ってデータを取得することが多く、BQテーブルそのものを取得することはあまりありません。BigQuery Storage APIでは、シンプルな行フィルタは適用できるものの、複雑なフィルタリングを適用する場合はクエリを使って取得する必要があります。
そのため、ここではBigQuery Storage APIのクライアントライブラリを直接使ったデータ取得は比較の対象から外します。

取得するデータは、BigQueryのパブリックデータからWikipediaのページビューのログデータのうち、2019年1月1日0時0分のデータとします。

bigquery-public-data.wikipedia.pageviews_2019

SELECT * FROM `bigquery-public-data.wikipedia.pageviews_2019` 
WHERE datehour = '2019-01-01'

カラム数は4ですが、これだけで5,287,235行もある、なかなか骨のあるデータです。

(100万行オーダのデータは分析タスクではザラにあるボリュームなのでちょうどいい検証対象かも)

比較対象

実行環境はcolaboratory上とします。

マジックコマンド

マジックコマンドを使った取得は以下のようになります。

%%bigquery --project <project_id> df_temp

SELECT * FROM `bigquery-public-data.wikipedia.pageviews_2019` 
where datehour = '2019-01-01'

colaboratoryのセルで上記のコマンドを実行することで、df_tempに取得結果がpandas.DataFrameとして得られます。

BigQuery クライアントライブラリ

colaboratoryではクライアントライブラリはすでにインストールされている状態なので、そのままインポートできます。

from google.cloud import bigquery

query = """
SELECT * FROM `bigquery-public-data.wikipedia.pageviews_2019` 
where datehour = '2019-01-01'
"""

client = bigquery.Client(project=<project_id>)
df_temp = client.query(query).to_dataframe()

BigQuery → (AVRO) → GCS

BigQuery Storage API Overviewで説明されている、2番目の方法に相当する手法です。

BQクライアントライブラリを使ってクエリ結果を一時テーブルとして保存し、GCSにAVROファイルとしてエクスポート、それを実行環境上にpandas.DataFrameとしてロードする、という方法です。

実験では、この方法でデータを取得する社内ライブラリを開発していたので、これを利用します。

マジックコマンド + BigQuery Storage API

BigQuery Storage APIをマジックコマンドで利用するには、以下のようにcontext.use_bqstorage_api プロパティを True に設定します。

import google.cloud.bigquery.magics
google.cloud.bigquery.magics.context.use_bqstorage_api = True

これを実行してから、先ほどのマジックコマンドを利用します。

%%bigquery --project <project_id> df_temp

SELECT * FROM `bigquery-public-data.wikipedia.pageviews_2019` 
where datehour = '2019-01-01'

BigQuery クライアントライブラリ + BigQuery Storage API

クライアントライブラリでBigQuery Storage APIを利用するには、to_dataframe()メソッドの引数にBigQuery Storage APIのクライアントオブジェクトを渡してあげるだけです。(ｶﾝﾀﾝ!)

from google.cloud import bigquery
from google.cloud import bigquery_storage_v1beta1

query = """
SELECT * FROM `bigquery-public-data.wikipedia.pageviews_2019` 
where datehour = '2019-01-01'
"""

bq_client = bigquery.Client(project=<project_id>)
bqstorage_client = bigquery_storage_v1beta1.BigQueryStorageClient()
df_temp = bq_client.query(query).to_dataframe(bqstorage_client)

実験

上で記したように、クエリからpandas.DataFrameまでの時間を計測します。
時間計測はマジックコマンド %time を該当する行に付けて計測します。
クライアントライブラリを使った場合は以下のような感じです::

from google.cloud import bigquery

query = """
SELECT * FROM `bigquery-public-data.wikipedia.pageviews_2019` 
where datehour = '2019-01-01'
"""

client = bigquery.Client(project=<project_id>)
%time df_temp = client.query(query).to_dataframe() # <-- ここを計測

マジックコマンド、マジックコマンド + BigQuery Storage APIでは、セルの1行目に%%timeを追記してセル全体の時間を計測します。

%%time
%%bigquery --project <project_id> df_temp

SELECT * FROM `bigquery-public-data.wikipedia.pageviews_2019` 
where datehour = '2019-01-01'

どちらもワンショットの計測なので、ばらつきがあるのは承知の上で計測します。

計測結果

手法	CPU time	Wall time
マジックコマンド	2min 9s	4min 48s
マジックコマンド + BigQuery Storage API	19.6 s	21.2 s
BQクライアントライブラリ	2min 9s	5min 1s
BQ -> (AVRO) -> GCS	57.1 s	1min 42s
BQクライアントライブラリ + BigQuery Storage API	19.7 s	36.8 s

結果より、BigQuery Storage APIを使った手法が圧倒的に速い。。。
桁違いの速さ。。

(BQ -> (AVRO) -> GCS はせっかくライブラリ作ったのになぁ。。)

おわりに

BQテーブルのデータをクエリからpandas.DataFrameとして取得する方法は、速度優先ならBigQuery Storage API一択という結論です。

最後に注意点。

2019/12/17時点ではまだベータ版なので、USとEUのマルチリージョンのみで利用できるそうです。
over viewにはUS/EUマルチリージョン + いくつかのロケーション(asia-east1, asia-northeast1, asia-south1, asia-southeast1, europe-west2, northamerica-northeast1)が含まれてますが、料金ページではUS/EUマルチリージョン以外はUnavailableと表示されています。
(実はドキュメントが追いついていないだけで使えたりする？)

利用する際はデータセットのロケーションにご注意を。。

ちなみに料金は、ReadRows の呼び出しによって BigQuery ストレージから読み取ったバイト数に基づいて料金を課金となっており、基本的にはBigQuery Storage APIのReadRowsで読み取ったデータ量に応じて課金される仕組みだそうです(認識違っていたらすみません)。

• クエリ実行する場合
  一時テーブルから読み取られたデータには料金はかかりません。とあるため、Storage API自体の料金は掛からず、 クエリ実行分の料金のみ発生。

• テーブル丸ごと取得する場合
  USマルチリージョンで$1.10 per TB

(@na0 ありがとうございます m(_ _)m)