SensorBeeに触れてみた


CentOSの環境を用意してGo言語もインストールしたので、SensorBeeというものを試してみます。

[WMwareでCentOS7環境を作る](http://qiita.com/estaro/items/23c10fe7e43a2a70e689)
[CentOSにGo言語のインストール](http://qiita.com/estaro/items/23c10fe7e43a2a70e689)

## SensorBeeとは何か

http://docs.sensorbee.io/en/latest/preface.html
>SensorBee is an open source, lightweight, stateful streaming data processing engine for the Internet of Things (IoT).

ひとことで言うと、「IoT向け軽量ストリーミングエンジン」ということらしいです。

## SensorBeeをインストールしてみる
goがインストールされていて、環境変数GOPATHが設定されている必要があります。

インストールは"go get"で行います。

```bash
$ go get gopkg.in/sensorbee/sensorbee.v0/...
```

## tutorialを動かしてみる

まずはtutorialを取得してきます。

```bash
$ go get github.com/sensorbee/tutorial/wordcount
```

設定ファイル類をワークディレクトリにコピーします。

```bash
$ mkdir -p work/sensorbee_tl
$ cp go/src/github.com/sensorbee/tutorial/wordcount/config/* work/sensorbee_tl/
```

コピーしたファイルは３つです。

```bash
$ cd work/sensorbee_tl/
$ ls
build.yaml  sensorbee.yaml  wordcount.bql
```

ファイルの内容を眺めてみます。

```yaml
$ cat build.yaml
plugins:
  - github.com/sensorbee/tutorial/wordcount/plugin
```

```yaml
$ cat sensorbee.yaml
topologies:
  wordcount:
    bql_file: wordcount.bql

network:
  listen_on: ":15601"
```

```sql
$ cat wordcount.bql
CREATE SOURCE sentences TYPE wc_sentences;

CREATE STREAM words AS
    SELECT RSTREAM name, text AS word
        FROM wc_tokenizer("sentences", "text") [RANGE 1 TUPLES];

CREATE STREAM word_counts AS
    SELECT ISTREAM word, count(*)
        FROM words [RANGE 60 SECONDS]
        GROUP BY word;
```

なんとなく雰囲気を感じますかね･･･？

続いて、build.yamlがあるディレクトリでbuild_sensorbeeを実行します。
すると「sensorbee」という実行ファイル、「sensorbee_main.go」というGoプログラムが作成されます

```bash
$ build_sensorbee
sensorbee_main.go
$ ls
build.yaml  sensorbee  sensorbee.yaml  sensorbee_main.go  wordcount.bql
```

「sensorbee_main.go」の中身は以下の通りとなります。

```go
$ cat sensorbee_main.go
package main

import (
 _ "github.com/sensorbee/tutorial/wordcount/plugin"
 _ "gopkg.in/sensorbee/sensorbee.v0/bql/udf/builtin"
 "gopkg.in/sensorbee/sensorbee.v0/cmd/lib/run"
 "gopkg.in/sensorbee/sensorbee.v0/cmd/lib/runfile"
 "gopkg.in/sensorbee/sensorbee.v0/cmd/lib/shell"
 "gopkg.in/sensorbee/sensorbee.v0/cmd/lib/topology"
 "gopkg.in/sensorbee/sensorbee.v0/version"
 "gopkg.in/urfave/cli.v1"
 "os"
 "time"
)

func init() {
 // TODO
 time.Local = time.UTC
}

func main() {
 app := cli.NewApp()
 app.Name = "sensorbee"
 app.Usage = "SensorBee built with build_sensorbee 0.7.1"
 app.Version = version.Version
 app.Commands = []cli.Command{
  run.SetUp(),
  runfile.SetUp(),
  shell.SetUp(),
  topology.SetUp(),
 }
 if err := app.Run(os.Args); err != nil {
  os.Exit(1)
 }
}
```

実行ファイル"sensorbee"を使用して、サーバーとして起動します。

```bash
$ ./sensorbee run
time="2017-08-26T08:58:26Z" level=info msg="Setting up the server context" config="{\"logging\":{\"log_destinationless_tuples\":false,\"log_dropped_tuples\":false,\"min_log_level\":\"info\",\"summarize_dropped_tuples\":false,\"target\":\"stderr\"},\"network\":{\"listen_on\":\":15601\"},\"storage\":{\"uds\":{\"params\":{},\"type\":\"in_memory\"}},\"topologies\":{}}"
time="2017-08-26T08:58:26Z" level=info msg="Starting the server on :15601"
```

続けて別のターミナルから操作します。

```bash
$ ss -nat | grep 15601
LISTEN     0      128         :::15601                   :::*
$ curl http://localhost:15601/api/v1/runtime_status
{"gomaxprocs":2,"goroot":"/usr/local/go","goversion":"go1.9","hostname":"monaca","num_cgo_call":1,"num_cpu":2,"num_goroutine":4,"pid":26461,"user":"centos","working_directory":"/home/centos/work/sensorbee_tl"}
```

Topologyというものを作ります。

```bash
$ ./sensorbee topology create wordcount
$ echo $?
0
```

作成したTopologyを指定してシェルを起動します。

```bash
$ ./sensorbee shell -t wordcount
wordcount> EVAL "Hello" || ", world!";
"Hello, world!"
```

"Topology"というものは後で調べた方がよさそうなキーワードですね。

ここで実行できるのがSQLライクな操作言語の**BQL**です。
BQLでSOURCEを作成します。

```sql
CREATE SOURCE sentences TYPE wc_sentences;
```

Tutorial通り、作成したSOURCEをFROM句に指定してSELECTします。

```sql
SELECT RSTREAM * FROM sentences [RANGE 1 TUPLES];
{"name":"jacob","text":"excepteur esse excepteur minim aliquip consectetur incididunt ipsum"}
{"name":"jacob","text":"do ut quis anim laboris officia elit mollit ipsum"}
{"name":"jacob","text":"ut nostrud sunt voluptate ad aliquip deserunt pariatur anim"}
```

FROM句にユーザ関数を適用してテキストを分割するサンプルです。

```sql
SELECT RSTREAM * FROM wc_tokenizer("sentences", "text") [RANGE 1 TUPLES];
{"name":"isabella","text":"ut"}
{"name":"isabella","text":"do"}
{"name":"isabella","text":"quis"}
{"name":"isabella","text":"enim"}
{"name":"isabella","text":"enim"}
{"name":"isabella","text":"pariatur"}
{"name":"isabella","text":"reprehenderit"}
{"name":"isabella","text":"enim"}
{"name":"isabella","text":"mollit"}
```

SQLのVIEWのようにSELECT結果を新たにSTREAMとすることもできます。

```sql
CREATE STREAM words AS 
  SELECT RSTREAM name, text AS word
    FROM wc_tokenizer("sentences", "text") [RANGE 1 TUPLES];
```

これでwordcountできるようになります。

```sql
SELECT ISTREAM word, count(*) FROM words [RANGE 60 SECONDS] GROUP BY word;
```

シェルの終了はexitです。

```sql
wordcount> exit
$
```

exitしてもサーバを停止しない限り、CREATEしたSTREAMなどは有効ですが、サーバを再起動すると、Topologyから失われます。
また、サーバが停止されていても、"./sensorbee shell -t wordcount"で対話シェルに入れます（なにかしようとすると当然エラーとなりますが）。

サーバを停止するごとにTopologyを作成しなおすのは大変なので、サーバ起動時に設定ファイルを指定することができます。

```bash
$ cat sensorbee.yaml
topologies:
  wordcount:
    bql_file: wordcount.bql

network:
  listen_on: ":15601"
$ ./sensorbee run -c sensorbee.yaml
```

wordcountのTutorialを一通り動かしてみたので、次はwordcountの内容を探ってみたいと思います。

終わりです。