3r3-31. Have you ever been asked to calculate the amount of something based on the data in the database for the last month, grouping the result by some values ​​and breaking it all by day /hour? 3r3408.  
If so, then you already have the idea that you have to write something like this, only worse than 3r3408.  
3r3408.  
SELECT hour (datetime), somename, count (*), sum (somemetric)
from table
where datetime>: monthAgo
group by ? 2
order by 1 desc, 2
3r3408.  
From time to time, a wide variety of such requests begin to appear, and if you can endure and help once, alas, applications will come in the future as well.  
And such requests are bad because they take the system resources well for the duration of the execution, and there can be so much data that even a replica for such requests would be a pity (and its time) 3r3408.  
3r3408.  
And what if I say that you can create a view directly in PostgreSQL that, on the fly, will take into account only new incoming data in a directly similar query, as above? 3r3408.  
So - this is how the PipelineDB
extension can do.  
3r3408.  
3r33383.
Demo from their site, how it works [/b]
3r3334. 3r3408.  
3r3408.  
Continuous Views and by default materialize, i.e. with preservation of the state 3r3408.  
In the expression 3r3188. WITH additional parameters are passed. 3r3408.  
In my case 3r3188. ttl = '3 month' suggests that you only need to store data for the last 3 months, and take the date /time from column 3r3188. M . Background process 3r3188. reaper looks for obsolete data and deletes it
 
For those who do not know - function 3r3188. minute returns date /time without seconds. Thus, all events that occurred in one minute will have the same time as a result of the aggregation of 3r33232.  
3r33434. Such a view is practically a table, because the index by date for the sample would be useful if there is a lot of data stored 3r3333368. create index on viewflow (m desc, action);
3r33434.  
3r-33199. 3r3408.  
3r33394. Using PipelineDB
3r3408.  
Remember: insert data into the stream, and read from the subscribed views of 3r3408.  
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 ()
3r3408.  
3r33383.
I execute the query manually [/b]
3r3408.  
First I observe how the data changes in the 46th minute
 
As soon as the 47th comes, the previous one stops updating and the current minute 3r3408 starts to tick.  
3r3408.  
If you look at the query plan, you can see the original table with the data 3r3408.  
3r3408.  
3r3408.  
3r3408.  
I recommend to go into it and find out how your data is actually stored 3r3408.  
3r3408.  
3r33383.
Event generator on C # [/b]
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 (5? 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); 3r3446.} {
Return _actions[_rnd.Next(_actions.Length)];
}
}
}
3r3408.  
3r3408.  
3r33394. Conclusion in Grafana
3r3408.  
To retrieve data from postgres, add the appropriate data source:
 
3r3408.  
3r3408.  
3r3408.  
Create a new dashboard and add a Graph type panel to it, and then go to the editing panel: 3r3408.  
3r3408.  
3r33333. 3r3408.  
3r3408.  
Next - select the data source, switch to the writing mode of the sql query and enter this:
select
m as time, - Grafana requires a time column
count, action
from viewflow
where $ __ timeFilter (m) is a macro of graphs, accepts a column name as input, output between col: startdate and: enddate
order by m desc, action;
3r3408.  
And then we get a normal schedule, of course, if you run the event generator 3r3408.  
3r3408.  
3r33333. 3r3408.  
3r3408.  
FYI: having an index can be very important. Although its use depends on the volume of the resulting table. If you plan to store a small number of lines for a small amount of time, it may very easily be that seq scan will be cheaper, and the index will only add extra. load when updating values ​​3r3408.  
3r3408.  
Multiple views of
can be signed to one stream.  
Suppose I want to see how many api methods are performed, broken down by percentiles 3r3408.  
CREATE VIEW viewflow_per WITH (ttl = '3 d', ttl_column = 'm') AS
select minute (dtmsk) m,
action,
percentile_cont (???) WITHIN GROUP (ORDER BY duration) :: smallint p5?
percentile_cont (???) WITHIN GROUP (ORDER BY duration) :: smallint p9?
percentile_cont (???) WITHIN GROUP (ORDER BY duration) :: smallint p99
from flow_stream
group by ? 2;
create index on viewflow_per (m desc);
3r3408.  
3r33383.
I am doing the same trick with graphena and get: 3r3333385.
3r3408.  
3r3408.  
3r33394.
total. 3r3408.  
In general, the thing is working, behaved well, without complaints. Although, under the docker, downloading their demo database in the archive (2.3 GB) turned out to be a little long case 3r3408.  
I want to note - I did not conduct stress tests 3r3408.  
Official documentation 3r3402. http://docs.pipelinedb.com/index.html
3r3408.  
3r3408.  
May be interesting 3r3408.  
3r33410.  
3r33434. Support data download 3r31313. from Apache Kafka to stream
3r33434.  
3r33434. Similarly, with 3r33418. Amazon Kinesis
3r33434.  
3r33434. Only can be created. for data transformation (without storage)
 
3r33434. 3r33434. PipelineDB Cluster
- there is a commercial version. In it it is possible to distribute views on shards. Read more in the dock on the cluster solution 3r33434.  
! function (e) {function t (t, n) {if (! (n in e)) {for (var r, a = e.document, i = a.scripts, o = i.length; o-- ;) if (-1! == i[o].src.indexOf (t)) {r = i[o]; break} if (! r) {r = a.createElement ("script"), r.type = "text /jаvascript", r.async =! ? r.defer =! ? r.src = t, r.charset = "UTF-8"; var d = function () {var e = a.getElementsByTagName ("script")[0]; e.parentNode.insertBefore (r, e)}; "[object Opera]" == e.opera? a.addEventListener? a.addEventListener ("DOMContentLoaded", d,! 1): e.attachEvent ("onload", d ): d ()}}} t ("//mediator.mail.ru/script/2820404/"""_mediator") () (); 3r33440.