PostgreSQL：PipelineDB-リアルタイム集約クエリ

SELECT hour(datetime), somename, count(*), sum(somemetric) from table where datetime > :monthAgo group by 1, 2 order by 1 desc, 2 

サイトのデモの仕組み

PipegreDBをPostgresにインストールする

1. apt update
2. apt install curl
3. curl -s http://download.pipelinedb.com/apt.sh | bash
4. apt install pipelinedb-postgresql-11
5. cd /var/lib/postgresql/data
6. 任意のエディターでpostgresql.confファイルを開きます
7. shared_preload_librariesキーを見つけ、コメントを外して、 pipelinedb値を設定します
8. キーmax_worker_processesは128に設定されています（推奨ドック）
9. サーバーを再起動

PipelineDBでストリームとビューを作成する

再起動後pg-そのようなことがあるようにログを監視します

1. 作業するデータベース： CREATE DATABASE testpipe;
2. 拡張機能の作成： CREATE EXTENSION pipelinedb;
3. 今、最も興味深いのはストリームを作成することです。 その中で、さらに処理するためにデータを追加する必要があります。
   
   

    CREATE FOREIGN TABLE flow_stream ( dtmsk timestamp without time zone, action text, duration smallint ) SERVER pipelinedb; 

   
   実際、通常のテーブルを作成するのと非常によく似ていselect単純なselectこのストリームからデータを取得することはできません-ビューが必要です
4. 実際にそれを作成する方法：
   
   

    CREATE VIEW viewflow WITH (ttl = '3 month', ttl_column = 'm') AS select minute(dtmsk) m, action, count(*), avg(duration)::smallint, min(duration), max(duration) from flow_stream group by 1, 2; 

   
   これらは連続ビューと呼ばれ、デフォルトで具体化されます。 状態の保存。
   
   WITHは追加のパラメーターを渡します。
   
   私の場合、 ttl = '3 month'は、過去3か月間のデータのみを保存し、 M列から日付/時刻を取得する必要があることを意味しますM バックグラウンドreaperプロセスreaper 、古いデータをreaperして削除します。
   
   知らない人のために、 minute関数は秒なしで日付/時刻を返します。 したがって、1分間に発生したすべてのイベントは、集約の結果として同じ時間になります。
5. 多くのデータが保存されている場合、サンプリングの日付によるインデックスが役立つため、このようなビューはほとんどテーブルです
   
   

    create index on viewflow (m desc, action); 

PipelineDBを使用する

 insert into flow_stream VALUES (now(), 'act1', 21); insert into flow_stream VALUES (now(), 'act2', 33); select * from viewflow order by m desc, action limit 4; select now() 

リクエストを手動で実行します

 using Npgsql; using System; using System.Threading; namespace PipelineDbLogGenerator { class Program { private static Random _rnd = new Random(); private static string[] _actions = new string[] { "foo", "bar", "yep", "goal", "ano" }; static void Main(string[] args) { var connString = "Host=localhost;port=5432;Username=postgres;Database=testpipe"; using (var conn = new NpgsqlConnection(connString)) { conn.Open(); while (true) { var dt = DateTime.UtcNow; using (var cmd = new NpgsqlCommand()) { var act = GetAction(); cmd.Connection = conn; cmd.CommandText = "INSERT INTO flow_stream VALUES (@dtmsk, @action, @duration)"; cmd.Parameters.AddWithValue("dtmsk", dt); cmd.Parameters.AddWithValue("action", act); cmd.Parameters.AddWithValue("duration", GetDuration(act)); var res = cmd.ExecuteNonQuery(); Console.WriteLine($"{res} {dt}"); } Thread.Sleep(_rnd.Next(50, 230)); } } } private static int GetDuration(string act) { var c = 0; for (int i = 0; i < act.Length; i++) { c += act[i]; } return _rnd.Next(c); } private static string GetAction() { return _actions[_rnd.Next(_actions.Length)]; } } } 

Grafanaでの結論

 select m as time, -- Grafana   time count, action from viewflow where $__timeFilter(m) --  ,     ,   col between :startdate and :enddate order by m desc, action; 

 CREATE VIEW viewflow_per WITH (ttl = '3 d', ttl_column = 'm') AS select minute(dtmsk) m, action, percentile_cont(0.50) WITHIN GROUP (ORDER BY duration)::smallint p50, percentile_cont(0.95) WITHIN GROUP (ORDER BY duration)::smallint p95, percentile_cont(0.99) WITHIN GROUP (ORDER BY duration)::smallint p99 from flow_stream group by 1, 2; create index on viewflow_per (m desc); 

私はgrafanaで同じトリックを行い、以下を取得します。

合計

面白いかもしれません

• Apache Kafkaからストリームへのデータのダウンロードのサポート
• Amazon Kinesisに類似
• ビューは、データ変換のみのために作成できます（ストレージなし）
• PipelineDBクラスター -商用バージョンがあります。 その中で、シャード間でビューを分散できます。 クラスター決定ドックの詳細