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クラスター -商用バージョンがあります。 その中で、シャード間でビューを分散できます。 クラスター決定ドックの詳細