Python3 (Jupyter Notebook) で Google BigQuery を操作し parquet 形式で保存 #機械学習

概要

データ分析を行う上で Python3 で Google BigQuery を操作することが多いです．
操作には下記の 2 つの手法が存在します．

Python の BigQuery 連携ライブラリを使う
Google Cloud Datalab を使う

今回は, Python の BigQuery 連携ライブラリを使う手法の内 pandas.io.gbq を使う手法について説明します．

今回の目標

BigQuery などのオンプレミス環境においてクエリを叩く回数を減らし, コストを削減することおよび, ローカル環境を圧迫しないようなデータの保持を行うことを目標にします. そのため Apache Parquet 形式で取ってきたデータを保存することにしました.

Apache Parquet 形式とは

csvなどの行志向のデータフォーマットと違い、列志向のフォーマットで、列単位でデータを取り出す分析用途に向いています。この記事で他の形式との比較をしています．特に大規模データファイルを圧縮する際に効率が良くなるのでビッグデータの分析データの保存に用いられます．

pandas.io.gbq ことはじめ

自分ははじめにこの記事を参考にして, pandas.io.gbq を使おうとしていました.


import pandas as pd
query = 'SELECT * FROM tablename'
data_frame = pd.read_gbq(query, 'project名')

すると, 下記のエラーが発生してしまいます.

GenericGBQException: Reason: 400 Invalid table name: `project.dataset.table`
[Try using standard SQL (https://cloud.google.com/bigquery/docs/reference/standard-sql/enabling-standard-sql)].

おそらく 2016 年の記事なので, 2019 年の仕様と異なってしまうことがエラーの原因であると考えられます.
さらにエラー文を読んでみると StandardSQL に設定できていないことが考えられるわけです.

Standard SQLとLegacy SQL

BigQueryで古くに使われていたのがLegacy SQLで現在使われているのがStandard SQLである. そのため, BigQueryに対して, どちらを使うかを指定してあげる必要がある.

操作手法

以下の方法で Python から Google BigQuery を操作します.この方式を利用することにより, オンプレミス環境においてクエリを叩く回数を減らしコストを削減可能で, Parquetファイルとgzipを利用し圧縮効率をあげローカルに落とすデータの容量を削減することが可能です．

ローカルに .parquet ファイル(データを保存したファイル)がないとき
1. ローカル .sql ファイルを読み込む
2. 読み込んだクエリで, BigQuery を叩く
3. データフレームに格納する
4. 列志向形式 (Apache Parquet) に圧縮して, .parquet 形式でローカルに保存する
ローカルに .parquet ファイルがあるとき
1. ローカルの .parquet ファイルを読み込む
2. データフレームに格納する

コード説明

自分は下記のようなコードを書いて, 上の条件を満たすようにしました.

test.pynb

import os 
from google.cloud import bigquery
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

#parquetを入手，ある場合は取得しない
filePath = "../input/test.parquet"
sqlPath = "../script/test.sql"
if(not os.path.isfile(filePath)): #1
    with open(sqlPath, 'r') as fd: 
        query = fd.read() #2
    df = pd.read_gbq(query, 'project名', dialect='standard') #3
    table = pa.Table.from_pandas(df) #4
    pq.write_table(table, filePath, compression='gzip') #5
else:
    table = pq.read_table(filePath) #6
    df = table.to_pandas() #7

軽く説明すると,

#1 では os ライブラリを用いて指定したパスにデータファイルがあるかないかを確認し, なかったら if 文内の処理に移行します
#2 では #1 で指定されたパスの .sql ファイルから, クエリを変数 query に代入します．
#3 では #2 で取得したクエリを BigQuery 上で実行します. またここでstandard-sqlにて記述されているクエリであることを指定してあげます.
#4 では #3 で実行したクエリの結果をデータフレーム型に代入します．
#5 では #4 で作ったデータフレームを parquet ファイルで gzip 形式に圧縮し保存します．
#6 では指定されたパスにデータファイルがあったとき, table にロードします
#7 では #6 でロードした table をデータフレーム型に変更します

実行結果

これを実行すると Jupyter Notebook 上で, 下記のような画面が出力されます.

リンクを踏み, 認証を行うことで, 認証コードが発行されます. そのコードを Jupyter Notebook 上のフォームに入力することで, Python3 の PandasGBQ で BigQuery を操作することができるようになります.

まとめ

以上より, Python を用いてGoogle BigQuery を操作することができます．また上記のようなコードを書くことで, コストを削減し, 圧縮効率をあげ, 効率よくデータを取ってくることが可能です.