前回は、PipelineDBが何者かざっくりと書いてみましたが、今回はひとまずPipelineDBをセットアップしてサンプル・データを流し込むところまでをやってみましょう。

Installation

参考： http://docs.pipelinedb.com/installation.html

Linuxの場合

PipelineDBのインストールは主要なLinuxOSそれぞれに対してパッケージが用意されているのですぐに完了します。
AmazonLinuxの場合であれば、

$ wget https://www.pipelinedb.com/download/0.8.4/amazon-linux -O pipelinedb-0.8.4.rpm
$ sudo rpm -ivh pipelinedb-0.8.4.rpm

上記でインストール完了です。インストール後、データの保存場所の初期化のため、下記を実行して下さい。

$ pipeline-init -D <データを保存したいDirectory>

Macの場合

Macで使いたい方はダウンロードページよりpkgファイルを落としてダブルクリックするだけで利用開始できます。

Dockerの場合

Dockerでもイメージが公開されているため、そちらでの利用も可能です。


$ docker run -v /dev/shm:/dev/shm pipelinedb/pipelinedb

起動・停止

起動・停止はpipline-ctlコマンドで実行します。

$ pipeline-ctl -D <データ保存Directory> -l <ログファイル名> start
$ cat <ログファイル名>
    ____  _            ___            ____  ____
   / __ \(_)___  ___  / (_)___  ___  / __ \/ __ )
  / /_/ / / __ \/ _ \/ / / __ \/ _ \/ / / / __  |
 / ____/ / /_/ /  __/ / / / / /  __/ /_/ / /_/ /
/_/   /_/ .___/\___/_/_/_/ /_/\___/_____/_____/
       /_/

LOG:  database system was shut down at 2015-10-28 00:17:51 JST
LOG:  MultiXact member wraparound protections are now enabled
LOG:  autovacuum launcher started
LOG:  database system is ready to accept connections
LOG:  continuous query scheduler started
LOG:  registering background worker "worker0 [pipeline]"
LOG:  registering background worker "combiner0 [pipeline]"
LOG:  starting background worker process "combiner0 [pipeline]"
LOG:  starting background worker process "worker0 [pipeline]"
LOG:  continuous query process "combiner0 [pipeline]" running with pid 65289
LOG:  continuous query process "worker0 [pipeline]" running with pid 65290

停止時はstopを指定するだけです。

$ pipeline-ctl -D pipelinedb stop
waiting for server to shut down.... done
server stopped

Create Continuous View

PipelineDB上のデータストリームはContinuousViewという形で表現されますが、これはSQLとして扱う上ではテーブルと考えてもらって結構です。
ひとまず今回は後述するサンプルデータに合わせたContinuousViewを設定します。

$ pipline pipeline
pipeline (9.4.4)
Type "help" for help.

pipeline=# CREATE CONTINUOUS VIEW wiki_stats AS
pipeline-# SELECT hour::timestamp, project::text,
pipeline-#         count(*) AS total_pages,
pipeline-#         sum(view_count::bigint) AS total_views,
pipeline-#         min(view_count) AS min_views,
pipeline-#         max(view_count) AS max_views,
pipeline-#         avg(view_count) AS avg_views,
pipeline-#         percentile_cont(0.99) WITHIN GROUP (ORDER BY view_count) AS p99_views,
pipeline-#         sum(size::bigint) AS total_bytes_served
pipeline-# FROM wiki_stream
pipeline-# GROUP BY hour, project;
CREATE CONTINUOUS VIEW

ContinuousView自体の解説はまた別な投稿にしたいと思いますが、ざっくり言うとストリームデータを一定条件でサマライズしたデータを保持するテーブルです。FROM句の部分のwiki_streamはstreamとよばれ、実際のストリームデータ一行一行と対応するテーブルのようなものになります。これは事前に定義する必要自体はありません（実際は事前に型定義した方がパフォーマンスがでるとのことですが）。このstreamにはINSERTやCOPYコマンドを通じてデータを投入していきます。

Sample Data

PipelineDBの公式がサンプルデータとしてWikipediaのページビューデータを利用しており、こちらを利用することである程度簡単にストリーム処理の中身を体感していただけるようになっています。
参考: http://dumps.wikimedia.org/other/pagecounts-raw/

先ほど作成したCONTINUOUS VIEWで元にしているwiki_streamというstreamにこれを流してみましょう。

$ curl -sL http://pipelinedb.com/data/wiki-pagecounts | gunzip | \
        psql -h localhost -p 5432 -d pipeline -c "
        COPY wiki_stream (hour, project, title, view_count, size) FROM STDIN"

ちなみにPostgreSQL互換なのでpsqlコマンドからもアクセス可能です。
サンプルデータ => 標準出力 => COPYコマンドという流れで、wiki_streamというstreamにタブ区切りのデータを流し込んでいます。

Query

いよいよクエリを流してみましょう。
またpipelineコマンドで繋いでみます。

$ pipeline pipeline
pipeline (9.4.4)
Type "help" for help.

pipeline=# select * from wiki_stats ORDER BY total_views DESC limit 3;
        hour         | project | total_pages | total_views | min_views | max_views |      avg_views       |    p99_views     | total_bytes_served
---------------------+---------+-------------+-------------+-----------+-----------+----------------------+------------------+--------------------
 2015-06-01 01:00:00 | en      |     2590639 |     7640042 |         1 |    368264 |   2.9490955706294856 | 28.8838758913487 |       247434016274
 2015-06-01 00:00:00 | en      |     2556148 |     7554038 |         1 |    406121 |   2.9552428106666750 | 29.1063268187446 |       243707663997
 2015-06-01 01:00:00 | en.mw   |           1 |     5560465 |   5560465 |   5560465 | 5560465.000000000000 |          5560465 |       143619712266
(3 rows)

pipeline=# select * from wiki_stats ORDER BY total_views DESC limit 3;
        hour         | project | total_pages | total_views | min_views | max_views |     avg_views      |    p99_views     | total_bytes_served
---------------------+---------+-------------+-------------+-----------+-----------+--------------------+------------------+--------------------
 2015-06-01 02:00:00 | en      |     2602704 |     7871005 |         1 |    371626 | 3.0241644843209216 | 29.2771876395043 |       255541204297
 2015-06-01 01:00:00 | en      |     2590639 |     7640042 |         1 |    368264 | 2.9490955706294856 | 28.8838758913487 |       247434016274
 2015-06-01 00:00:00 | en      |     2556148 |     7554038 |         1 |    406121 | 2.9552428106666750 | 29.1063268187446 |       243707663997
(3 rows)

pipeline=# select count(*) from wiki_stats;
 count
-------
  3970
(1 row)

データを流しながら簡単に同じクエリを何度かselectしてたり、全行数をカウントしてみましたが、実際に流しているアクセスログの行数に対して、保持しているデータサイズは圧倒的に少ない状態です。
ここまで来るとなんとなくPipelineDBのContinuousViewを通したストリーム集計についてわかってくるのではないでしょうか。

最後に

PipelineDB自体はセットアップも簡単ですし、ひとまず触れて見つつStream処理の中身に触れてみてはいかがでしょうか。

次回は、ContinuousViewの解説 or fluentdからPipelineDBへ接続しつつデータを流してみるところまでやってみます。