BigQuery Storage API を触ってみた

Google Cloud Next '19で発表された新機能を紹介します！　(Cloud Run, BigQuery Storage API, Cloud Data Fusion) - ZOZO Technologies TECH BLOG の記事を読んで、BigQuery Strage APIに興味が湧きました。APIの概要や、簡単に実験して分かったことをまとめました。

BigQuery Storage API の概要

• BigQueryからデータを取得するための、tabledata.list や jobs.getQueryResults のREST API method、extract jobs とは異なるアプローチを提供
• BigQuery(ストレージ層)からのデータの読み取りをストリーム処理で実行できる
  • データの格納はできない
• 複数のストリームから互いに独立した行のセットを読み取ることができる
• ストリームに対してlatencyを減らすようにデータを割り当ててくれるので、複雑な負荷分散ロジックが必要ない

ML分野ではとても役立ちそうです

Doc

簡単に触ってみるにはこの辺りを読むと良いと思います

• over view
  • https://cloud.google.com/bigquery/docs/reference/storage/
• product ドキュメント
  • BigQuery Storage API Overview  |  BigQuery  |  Google Cloud
• storage-api で取得したデータをpandasで受け取る
  • BigQuery Storage API を使用して BigQuery データを pandas にダウンロードする  |  BigQuery  |  Google Cloud
• source
  • google-cloud-python/bigquery_storage at master · googleapis/google-cloud-python
• python-client
  • Python Client for BigQuery Storage API (Beta) — google-cloud 470347f documentation

API使用の流れ

https://cloud.google.com/bigquery/docs/reference/storage/ の Basic API flow に記載があります。ここでは各処理を1行で説明。

• Create a session
  • 読み取り対象(table,columns等)の宣言をし、ストリームの要求をする

↓

• Read from a session stream
  • streaming RPCでデータを読み取る

↓

• Decode row blocks
  • Avro binary formatのRow blocksをデシリアライズする

APIを使用したデータ取得の実装に関して

BigQuery Storage API を使用して BigQuery データを pandas にダウンロードする  |  BigQuery  |  Google Cloud
もしくは
Python Client for BigQuery Storage API (Beta) — google-cloud 8b18324 documentation
に実装例があります。パラメータを設定する程度で、ほぼ実装例そのままで動きました。

速度検証してみた

テーブルの全データを読み出してPandasDataFrameに変換するにはどの程度の時間がかかるかを検証しました。(※ 環境、実装により速度は変わるので参考値です。)

検証条件

• 読み出し対象テーブル
  • データの場所:US
  • 50カラム
  • 1,357,662 レコード
  • 600Mbyte
• 読み出す側
  • 実装
    • BigQuery Storage API を使用して BigQuery データを pandas にダウンロードする  |  BigQuery  |  Google Cloud の BigQuery Storage API クライアント ライブラリを使用してテーブルデータをダウンロード とほぼ同じ実装です
  • 実行環境
    • 東京で私が所持するmac(local)で実行

結果

ストリームが1つしか確立されたない状況でしたが、テーブルの全データを読み出してPandasDataFrameに変換するには300秒程度でした。extract jobsでGCSにAVROに書き出して、GCSからlocalにダウンロードして、PandasDataFrameに変換する場合だと260秒だったので良い勝負が出来ているかと思います。(ストリーム処理で扱うには多いデータ量でしたが、さくっと試すにはこのデータしかなかった)

わかったこと

• 実装も複雑にならないし、データ読み出しの速さもあるので十分に活用できそう
• ストリーム数の調整が上手くできなかった
  • 下記パッケージドキュメントの解説通りにrequested_streamsを指定したが、単一ストリームになってしまう。requested_streamsを指定していない際に同じテーブル、Column Projectionでも複数のストリームが作成されたり、単一のストリームだったりと条件が理解できなかった。パッケージドキュメントのrequested_streamsのThe number of streams may be lower than the requested number, depending on the amount parallelism that is reasonable for the table and the maximum amount of parallelism allowed by the systemとあるが、requested_streamsには2~3程度して指定していないのに..
  • https://cloud.google.com/bigquery/docs/reference/storage/rpc/google.cloud.bigquery.storage.v1beta1#google.cloud.bigquery.storage.v1beta1.CreateReadSessionRequest

私の理解度だとこれが限度だったので、ストリーム数の調整方法やご指摘等ありましたらご連絡ください