Google App Engine for GoからBigQuery APIを使ってみる

注：古い記事の為、内容が最新ではない可能性がありますm(＿＿)m

どうもこんにちわ！マツウラです。
今回はGAE/Goと、BigQueryの組み合わせです。
BigQueryにデータセットを用意して、クエリを実行するところまでやっていこうと思います。

参考

新規プロジェクトの作成

BigQueryを使う前に、Google Developers Consoleからプロジェクトを作成します。
プロジェクトごとに提供されるIDはGoogleがプロジェクトを特定し、BigQuery APIの使用許可を貰うためにアプリで使います。

Google Developers Consoleで、新規プロジェクトを作成します。（使えるプロジェクトが既にあるならばOK）
サイドメニューのAPIS & AUTH -> APIsから、BigQuery APIをONに切り替えます。
課金がまだ有効でない場合は有効にしてください。

Google APIs Client Libraries

BigQuery APIの呼び出しはGoogle APIs client librariesを使います。
go getを使用して、Go言語用のライブラリをダウンロードしてください。 Google APIs Client Library for Go

go get code.google.com/p/google-api-go-client/bigquery/v2

goauth2

サービスに渡すHTTPクライアントは、リクエストにGoogleがサポートしている認証情報を自動的に追加する必要があります。なので、Go言語用のOAuth2.0クライアントライブラリであるgoauth2を使用します。

実際に使用するのはgoauth2の中のserviceaccountパッケージです。
serviceaccountパッケージは、サービスアカウントを使用してApp EngineからのOAuth認証リクエストの作成をサポートしてくれます。

go get code.google.com/p/goauth2/oauth
go get code.google.com/p/goauth2/appengine/serviceaccount

BigQueryにサンプルデータを用意する

BigQueryに今回のサンプルで取得するためのデータを準備します。

Google Cloud Consoleを開き、さきほど作成したプロジェクトのページを開きます。
左サイドのメニューから、BIG DATA -> BigQueryをクリックします。
プロジェクトごとのBigQueryの管理ページが開かれます。
左サイドのメニューに表示されているプロジェクト名の横、▼アイコンのドロップダウンメニュー -> Create new datasetをクリックします。
作成するデータセットのIDを入力します。（サンプルではtest_dataです）
左サイドのメニューに追加されたデータセットのドロップダウンメニュー -> Create new tableをクリックします。
Dataset IDに先ほどのデータセットID、Table IDに作成するテーブルのIDを入力します。
Source formatでJSONを選択し、Choose fileボタンをクリックしてJSONファイルを選択します。
（下記にサンプルとしてJSONファイルを載せておきます。）
SchemaにJSONデータに対応するスキーマを入力します。
下記JSONの場合は、kind:string,name:string,population:integer　となります。
Submitボタンをクリックします。

{"kind": "country", "name": "Shizuoka", "population": 707183}
{"kind": "country", "name": "Hamamatsu", "population": 791546}

サイドメニューから、テーブル名をクリックし、右上にあるDetailsボタンをクリックします。
無事にJSONデータが取り込まれていれば、データが表示されます。

テーブルデータの取得には、Google API Client Libraries、serviceaccountを使用します。

ライブラリの説明

Google API LibrariesはGo言語も含め、各言語ごとに自動生成されています。
そのため使用する言語において100％自然な書き方ではないように感じられるかもしれません。
Go言語版ではかなり自然な形になっていますが、ちょっとした非慣用句については目をつぶって欲しいとのことでした。

インストールしたライブラリの使い方は次のとおりです。

package main

import (
    "code.google.com/p/google-api-go-client/bigquery/v2"
)

パッケージ名はバージョン番号無しのAPI名になります。
上記の例では、bigqueryです。

各APIは∗http.Clientを引数にとるNew関数があり、API固有の∗Serviceを返します。
サービスを作成する例は次のとおりです。

service, err := bigquery.New(client)

OAuth HTTP Clientについて

サービスに渡すHTTPクライアントには、リクエストにGoogleがサポートする認証情報が自動的に追加されるようにする必要があります。
そこでgoauth2ライブラリを使用します。

また、今回はGoogle App Engineアプリから実行するため、OAuth2.0アクセストークンの生成にはappengineパッケージが提供するAccessToken()関数を使用します。

とはいっても、今回はこれらの設定を自分で用意する必要はありません。
goauth2ライブラリに含まれるserviceaccountパッケージは、サービスアカウントを使用してGoogle App EngineからOAuth2.0認証済みのHTTPクライアントリクエスト作成をサポートしてくれます。

serviceaccountパッケージの使い方は次のとおりです。

c := appengine.NewContext()
client, err := serviceaccount.NewClient(c, "https://www.googleapis.com/auth/bigquery")
if err != nil {
    c.Errorf("failed to create service account client: %q", err)
    return err
}

クエリの実行

それでは上記をふまえて、実際にBigQuery APIに指定したテーブルを取得するクエリリクエストを実行してみます。

import (
    "appengine"
    "code.google.com/p/goauth2/appengine/serviceaccount"
    "code.google.com/p/google-api-go-client/bigquery/v2"
    "fmt"
    "net/http"
)

const (
    // ３つとも各自で決めたIDに書き換えて下さい。
    PROJECT_ID string = "metal-bus-589"
    DATASET_ID string = "sample_dataset"
    TABLE_ID   string = "sample_table"
)

func init() {
    http.HandleFunc("/query", QueryHandler)
}

func QueryHandler(rw http.ResponseWriter, req *http.Request) {
    c := appengine.NewContext(req)
    client, err := serviceaccount.NewClient(c, bigquery.BigqueryScope)
    if err != nil {
        fmt.Fprintf(rw, "%s", err.Error())
        return
    }
    service, err := bigquery.New(client)
    if err != nil {
        fmt.Fprintf(rw, "%s", err.Error())
        return
    }

    response, err := service.Jobs.Query(PROJECT_ID, &bigquery.QueryRequest{
        Kind:  "bigquery#queryRequest",
        Query: "SELECT name FROM [" + DATASET_ID + "." + TABLE_ID + "] LIMIT 1000",
    }).Do()
    if err != nil {
        fmt.Fprintf(rw, "%s", err.Error())
        return
    }
 
    // response（テーブルデータ）を使用する。
    for _, row := range response.Rows {
        for _, cell := range row.F {
            fmt.Fprintf(rw, "%v\n", cell.V)
        }
    }
}

bigquery.Service.Jobs.Query()は、同期的にBigQuery SQL クエリを実行し、指定したタイムアウト時間内に完了した場合に結果を返します。
Query()関数には、プロジェクトIDおよび、リクエストのリソースタイプ、BigQueryクエリシンタックスに準ずるクエリ文字列が必要となります。

上記のサンプルでは、指定したテーブルからname列を上限1000行まで取得しています。

今回はライブラリを使用してBigQuery APIからクエリを実行してみました。
次回はStreaming insertを使用してリアルタイムにデータを追加してみたいと思います。

The Round

合同会社ナイツオの開発ブログ