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

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

そこで、[Google Cloud Next '19で発表された新機能を紹介します！　(Cloud Run, BigQuery Storage API, Cloud Data Fusion)](https://techblog.zozo.com/entry/google_cloud_next_19#BigQuery-Storage-API)で紹介されていた**BigQuery Storage API**でどれくらい時短になるのかを試してみます。

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

* [BigQuery Storage API Overview](https://cloud.google.com/bigquery/docs/reference/storage/)
* [BigQuery Storage API を使用して BigQuery データを pandas にダウンロードする](https://cloud.google.com/bigquery/docs/bigquery-storage-python-pandas?hl=ja)
* [Python Client for BigQuery Storage API (Beta)](https://googleapis.dev/python/bigquerystorage/latest/index.html)

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

```shell
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`

```sql
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ではクライアントライブラリはすでにインストールされている状態なので、そのままインポートできます。

```python
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](https://cloud.google.com/bigquery/docs/reference/storage/#background)で説明されている、2番目の方法に相当する手法です。

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

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

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

```python
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のクライアントオブジェクトを渡してあげるだけです。(ｶﾝﾀﾝ!)

```python
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` を該当する行に付けて計測します。
クライアントライブラリを使った場合は以下のような感じです::

```python
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](#マジックコマンド--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](#bigquery--avro--gcs) はせっかくライブラリ作ったのになぁ。。)

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

最後に注意点。

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

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

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

* クエリ実行する場合
    `一時テーブルから読み取られたデータには料金はかかりません。`とあるため、Storage API自体の料金は掛からず、 クエリ実行分の料金のみ発生。
    
* テーブル丸ごと取得する場合
    USマルチリージョンで$1.10 per TB


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