PythonからBigQueryを操作するときは BigQuery-Python が便利だった

Pythonから扱う場合いろいろな方法はあると思いますが、個人的にBigQuery-Pythonを使うことが多いので簡単にメモしておきます。

インストール

$ pip install bigquery-python

ドキュメント

準備

以下の情報を取得しておきます

• サービスアカウント
• キーファイル(p12 or pem)
• プロジェクト名

簡単な例

BigQueryClientの取得

from bigquery import get_client

PROJECT_ID = 'project-999'
SERVICE_ACCOUNT = '[e-mail]'
PRIVATE_KEY_PATH = '/path.to/keyfile.p12'

with open(PRIVATE_KEY_PATH, 'rb') as f:
    private_key = f.read()

client = get_client(PROJECT_ID,
                    private_key=private_key,
                    service_account= SERVICE_ACCOUNT,
                    readonly=False)

datasetの作成

DATASET = 'spam'

if not client.check_dataset(DATASET):
    client.create_dataset(DATASET)

tableの作成

schema.json

[
  {
    "name": "id",
    "type": "INTEGER"
  },
  {
    "name": "name",
    "type": "STRING"
  },
  {
    "name": "time",
    "type": "TIMESTAMP"
  }
]

DATASET = 'spam'
TABLE_NAME = 'egg'
SCHEMA_PATH = '/path.to/schema.json'

with open(SCHEMA_PATH, 'r') as f:
    table_schema = json.load(f)

if not client.check_dataset(DATASET):
    raise

if not client.check_table(DATASET, TABLE_NAME):
	client.create_table(DATASET, TABLE_NAME, table_schema)

(スキーマをjsonにする必要は無いですが使い回しが出来そうなので今回はjsonにしています。)

GoogleCloudStorageからのimport

//spam/egg.csv

"id", "name", "time"
1, "S", "2015-05-18 00:00:00"
2, "Y", "2015-11-02 00:00:00"

from bigquery import JOB_SOURCE_FORMAT_CSV

KEY_NAME = 'gs://spam/egg.csv'
TABLE_NAME = 'egg'

client.import_data_from_uris(
    TABLE_NAME,
    KEY_NAME,
    field_delimiter='\t',  # タブ区切り
    source_format=JOB_SOURCE_FORMAT_CSV,  # csv
    skip_leading_rows=1)  # csvの1行目をスキップ

queryの実行

同期的に取得する場合はtimeoutを指定します。それ以上時間がかかった場合はBigQueryTimeoutExceptionが発生します。

from bigquery.errors import BigQueryTimeoutException

query = 'SELECT id, name, time FROM [spam.egg] LIMIT 10'

try:
    job_id, results = client.query(query, timeout=60)
except BigQueryTimeoutException as e:
    print e

# job_id: u'job_xxxx_xxxxx'
# results: [
#    {u'id': 1, u'name': 'S',  u'time': "2015-11-02 00:00:00", }, 
#    {u'id': 2, u'name': 'Y',  u'time': "2015-11-02 00:00:00", }, 
# ]

時間がかかりそうな場合は非同期で取得することも検討できます。その場合はtimeoutを指定しません。

query = 'SELECT id, name, time FROM [spam.egg] LIMIT 10'
job_id, results = client.query(query)

# job_id: u'job_xxxx_xxxxx'
# results: []

job_idが発行されるので、get_query_rowsの引数にjob_idを入れると中身を取得できます。

completed, _total_rows = client.check_job(job_id)

if completed:
    results = client.get_query_rows(job_id)

# results: [
#    {u'id': 1, u'name': 'S',  u'time': "2015-11-02 00:00:00", }, 
#    {u'id': 2, u'name': 'Y',  u'time': "2015-11-02 00:00:00", }, 
# ]

その他

その他、BigQueryClientの中を見るといろいろ出来ることがわかります。

• check_dataset
• check_job
• check_table
• create_dataset
• create_table
• create_view
• dataset_resource
• delete_dataset
• delete_table
• export_data_to_uris
• get_dataset
• get_datasets
• get_query_results
• get_query_rows
• get_query_schema
• get_table
• get_table_schema
• get_tables
• import_data_from_uris
• patch_dataset
• push_rows
• query
• schema_from_record
• update_dataset
• wait_for_job
• write_to_table

Query Builder

render_queryを呼び出すといい感じにクエリを作ってくれる仕組みもあります。自分が使った感じだと、探し方が悪かったのかもしれませんがLIMITの追加やテーブルデコレータ、JSON_EXTRACTなどBigQuery独自の関数の扱いが出来なかったのでその辺りは自分で書かなければならなそうです。
https://github.com/tylertreat/BigQuery-Python#query-builder

GoogleのAPIドキュメント

ベースはGoogleのAPIなので状況に応じてGoogleのAPIドキュメントが役に立ちます。
https://cloud.google.com/bigquery/docs/reference/v2/

Linux環境での運用

ここに書かれている方法でキーファイルを扱う場合、Linux環境では暗号関連の扱いでエラーになる場合があります。BigQuery-Pythonをインストールする際に依存するPythonライブラリは同時にインストールされますが、Linux環境では依存するツールが無くてインストールされない場合があります。そのひとつがcryptographyというライブラリです。

BigQuery-Pythonの内部でImportErrorをキャッチして他のライブラリを使おうとしますが、それでも解決しないときはスクリプトの実行時にエラーになるので厄介です。cryptographyのインストール自体は下記でインストールできます。

$ pip install cryptography

しかし、そのために依存するツールをyumなどでインストールしなければなりません。ドキュメントに詳しく書かれていますが、自分の環境では以下を実行しました。

$ sudo yum install gcc libffi-devel python-devel openssl-devel

Cryptography ドキュメント
http://cryptography.readthedocs.org/en/latest/installation/