Big Sky

Treasure Data の REST API をコールするライブラリを書いた。

REST API | Treasure Data

REST API The user can control Treasure Data using the public REST API. This article will explain how...

http://docs.treasure-data.com/articles/rest-api

mattn/go-treasuredata ・ GitHub
https://github.com/mattn/go-treasuredata

使い方はこんな感じ。
client := treasuredata.NewClient("TREASUREDATA-API-KEY")
job, _ := client.JobIssueHive("mydb", "select * from mytable")
client.JobResultFunc(job.JobId, func(row []interface{}) error {
    fmt.Println(row)
    return nil
})
結果は JSON で貰えます。データが大量に返る場合があるのでデータ取得はコールバックで行います。error を返せば途中で抜けられます。
td フォルダにはコマンドラインツールが入っていて、クエリをコマンドラインから投げられます。
$ cd td
$ go build td.go
$ export TREASURE_DATA_API_KEY=xxxxxx
$ ./td -d mattn -q 'select * from unko'
["残念","小",{"size":"小","feel":"残念","time":"1363948092"},1363948092]
["残念","小",{"size":"小","feel":"残念","time":"1363948093"},1363948093]
["残念","小",{"size":"小","feel":"残念","time":"1363948094"},1363948094]
["残念","小",{"size":"小","feel":"残念","time":"1363948095"},1363948095]
["残念","小",{"size":"小","feel":"残念","time":"1363948096"},1363948096]
["残念","小",{"size":"小","feel":"残念","time":"1363948097"},1363948097]
["爽快","大",{"size":"大","feel":"爽快","time":"1363948098"},1363948098]
["残念","小",{"size":"小","feel":"残念","time":"1363948091"},1363948091]
["残念","小",{"size":"小","feel":"残念","time":"1363948092"},1363948092]
["残念","小",{"size":"小","feel":"残念","time":"1363948093"},1363948093]
["残念","小",{"size":"小","feel":"残念","time":"1363948094"},1363948094]
["残念","小",{"size":"小","feel":"残念","time":"1363948095"},1363948095]
["残念","小",{"size":"小","feel":"残念","time":"1363948096"},1363948096]
["残念","小",{"size":"小","feel":"残念","time":"1363948097"},1363948097]
["爽快","大",{"size":"大","feel":"爽快","time":"1363948098"},1363948098]
["残念","小",{"size":"小","feel":"残念","time":"1363948092"},1363948092]
["残念","小",{"size":"小","feel":"残念","time":"1363948093"},1363948093]
["残念","小",{"size":"小","feel":"残念","time":"1363948094"},1363948094]
["残念","小",{"size":"小","feel":"残念","time":"1363948095"},1363948095]
["残念","小",{"size":"小","feel":"残念","time":"1363948096"},1363948096]
["残念","小",{"size":"小","feel":"残念","time":"1363948097"},1363948097]
["爽快","大",{"size":"大","feel":"爽快","time":"1363948098"},1363948098]
["残念","小",{"size":"小","feel":"残念","time":"1363948091"},1363948091]
["残念","小",{"size":"小","feel":"残念","time":"1363948092"},1363948092]
["残念","小",{"size":"小","feel":"残念","time":"1363948093"},1363948093]
["残念","小",{"size":"小","feel":"残念","time":"1363948094"},1363948094]
["残念","小",{"size":"小","feel":"残念","time":"1363948095"},1363948095]
["残念","小",{"size":"小","feel":"残念","time":"1363948096"},1363948096]
["残念","小",{"size":"小","feel":"残念","time":"1363948097"},1363948097]
["爽快","大",{"size":"大","feel":"爽快","time":"1363948098"},1363948098]
毎日のウンコのサイズを treasure data に入れておいて、go-treasuredata で履歴を照会出来る様にしておけば健康管理もバッチリですね。

golang の特徴と言えば goroutine と channel ですが、その使いどころに悩む人もおられる様です。
goroutine は非同期に実行される処理、channel はその groutine と通信する為の仕組みと考えると分かりやすいです。
package main

import (
    "fmt"
    "time"
)

func main() {
    task := make(chan string)
    taskquit := make(chan bool)
    workerquit := make(chan bool)

    go func() {
    loop:
        for {
            select {
            case <-taskquit:
                workerquit <- true
                break loop
            case job := <-task:
                fmt.Println(job)
            }
        }
    }()

    go func() {
        for n := 0; n < 3; n++ {
            task <- fmt.Sprintf("お仕事%03d", n+1)
            time.Sleep(1 * time.Second)
        }
        taskquit <- true
    }()

    <-workerquit
}
この様に goroutine との間でキューの役割を果たします。channel はバッファを持っておりある程度の量ならばブロックする事無く goroutine と通信出来ます(バッファサイズはmakeで変更可能)。
この goroutine と channel をどの様に使うかですが、まず非同期を意識せずに主たる目的を持って処理を書きます。

• ある処理から「お仕事」が3回投げかけられる
• ある処理では「お仕事」の依頼を受けると逐次処理する

ここを非同期にする事でパフォーマンスアップを行います。速度面もそうですが上記の様な処理を順番に行う為には全ての「お仕事」をメモリに蓄積しておくか、「お仕事」の依頼があるたびにその処理を実行するコールバックを呼び出してやる必要があります。
しかし groutine と channel を使う事で

• お仕事の依頼を投げる側
• お仕事の依頼を受ける側

に処理を分割する事が出来ます。お仕事を投げる側はソケットか何かで電文を受信してパースし、チャネルに「お仕事」を投げつけます。
task <- fmt.Sprintf("お仕事%03d", n+1)
そしてお仕事を受ける側はその channel を指定して「お仕事」を受け取ります。
しかしお仕事を受ける側は、永遠にループする訳にはいけません。終了指示があれば中途半端にお仕事を処理せずに、グレースフルに終了したい物です。そこで終了用の channel を用意して、それを両方 select します。
例ではタスク登録 goroutine 終了用とワーカー終了用を用意しました。
select {
case <-taskquit:
    workerquit <- true
    break loop
case job := <-task:
    fmt.Println(job)
}
元々、別の関数で実装されていた処理というのは並行でない事はもちろんですが、多くのメモリを必要とします。さらにそれをコールバックベースの処理に作り替えると、大きくコードを壊さなければなりません。また純粋に並行ではありません。
golang の goroutine と channel を使えば、既存の処理を簡単に並行化する事が出来るのです。
今回の例では両方共に goroutine にしましたが、この目的で言えばお仕事を投げる側は goroutine でなくても構いません。
例えば Web でデータを登録し、それをバックグランドジョブとして後処理したいといった場合に、この channel の威力が発揮されるでしょう。

追記
先にワーカーが終了しないケースがあったのでコードを修正。

golang - Go言語における埋め込みによるインタフェースの部分実装パターン - Qiita [キータ]
http://qiita.com/tenntenn/items/e04441a40aeb9c31dbaf

golang はインタフェースがマッチしているかどうかにより処理を切り分けられる。 package main

import "fmt"

type Person struct {
    FirstName string
    LastName  string
}

func (p *Person) Name() string {
    return p.FirstName + " " + p.LastName
}

func main() {
    person := &Person{"Taro", "Yamada"}
    fmt.Println(person.Name())
}
メソッドを保持しているのであれば、インタフェースにアサーション出来る。
package main

import "fmt"

type Person struct {
    FirstName string
    LastName  string
}

func (p *Person) Name() string {
    return p.FirstName + " " + p.LastName
}

type Named interface {
    Name() string
}

func printName(named Named) {
    fmt.Println(named.Name())
}

func main() {
    person := &Person{"Tarou", "Yamada"}
    printName(person)
}
そして実は型がマッチしているかを確認する事も出来る。interface{} にキャストした後に指定のインタフェースで型アサーションを行う。その際に戻り値の2つめにアサーションが成功したかどうかが返される。 package main

import "fmt"

type Person struct {
    FirstName string
    LastName  string
}

func (p *Person) Name() string {
    return p.FirstName + " " + p.LastName
}

type Named interface {
    Name() string
}

// パーさんは Name メソッドを持たない
type Persan struct {
    FirstName string
    LastName  string
}

func printName(named Named) {
    fmt.Println(named.Name())
}

func main() {
    person := &Person{"Tarou", "Yamada"}
    named, ok := interface{}(person).(Named)
    if ok {
        printName(named)
    } else {
        fmt.Println("Person is not Named intreface")
    }

    persan := &Persan{"Tarou", "Yamada"}
    named, ok = interface{}(persan).(Named)
    if ok {
        printName(named)
    } else {
        fmt.Println("Persan is not Named intreface")
    }
}
つまり例えば、ライブラリやドライバが新しいメソッドに対応しているかどうか、と言った事を動的に確認出来る様になります。
この手法はオフィシャルパッケージの database/sql でも使われており、各ドライバが指定のインタフェースを実装しているかどうかを確認しています。 type Execer interface {
    Exec(query string, args []Value) (Result, error)
}
Exec というメソッドを持ったインタフェース Execer を用意して if execer, ok := dc.ci.(driver.Execer); ok {
}
Execer にアサーション可能な場合だけ特別な処理を行う、という事をやっています。
C/C++ の様に依存ライブラリの特定バージョン以降でメソッドがあったり無かったりという呪縛から解き放たれる訳です。
これがリフレクション無しに出来るのは便利ですね。