Изучив это еще немного, я думаю, что реальный ответ заключается в том, что не существует готового способа добиться этого эффективным образом. Тем более, что объем данных увеличивается. В конечном счете, объединение многих тысяч строк будет медленным, потому что это просто большой объем доступа к данным.
Вместо этого я пошел другим путем. Я использую триггер Postgresql для таблицы, в которой хранятся необработанные потоки (traffic_flow). Каждый раз, когда запись вставляется в traffic_flow, триггер будет вставлять новые данные в отдельные таблицы агрегации для ежедневных, почасовых и поминутных данных.
Вот моя экспериментальная реализация, вдруг кому пригодится. Это можно улучшить, чтобы также обрабатывать обновления и удаления.
create or replace function update_aggregated_traffic(NEW RECORD, table_name TEXT, interval_name text, store_customer BOOLEAN)
returns void
language plpgsql
as
$body$
declare
aggregate_interval interval;
customer_ip_ inet;
begin
-- Update the data aggregated traffic data given the insertion of a new flow.
-- A flow is the data about a single connection (start time, stop time, total
-- bytes/packets). This function essentially rasterises that data into a
-- series of aggregation buckets.
-- interval_name should be second, hour, or minute
-- turn the interval_name into an actual INTERVAL
aggregate_interval = ('1 ' || interval_name)::INTERVAL;
if store_customer then
customer_ip_ = NEW.source_address;
else
customer_ip_ = '100.64.0.0'::INET;
end if;
-- We need to insert into a dynamically generated table name. There is
-- no way to do this without writing the whole SQL statement as a string.
-- Instead, let's use a trick. Create a temporary view, then insert into that.
-- Postgres will proxy this insert into the desired table
drop view if exists table_pointer;
execute format('create temporary view table_pointer as select * from %s', table_name);
-- We use a CTE to keep things readable, even though it is pretty long
with aggregate_range AS (
-- Create all the aggregate buckets spanned by the inserted flow
SELECT generate_series(
date_trunc(interval_name, lower(NEW.range)),
date_trunc(interval_name, upper(NEW.range)),
aggregate_interval
) as range_lower
),
-- For each bucket, figure out its overlap with the provided flow data.
-- Only the first and last buckets will have less than than complete overlap,
-- but we do the calculation for all buckets anyway
with_overlaps AS (
SELECT
NEW.range * tstzrange(range_lower, range_lower + aggregate_interval) AS overlap,
range_lower
FROM
aggregate_range
),
-- Convert the overlap intervals into seconds (FLOAT)
with_overlap_seconds AS (
SELECT
extract(epoch from (upper(overlap) - lower(overlap))) as overlap_seconds,
range_lower
FROM
with_overlaps
)
-- Now we have enough information to do the inserts
insert into table_pointer as traffic
(timestamp, customer_ip, as_number, bytes, packets)
select
range_lower,
customer_ip_,
NEW.as_number,
-- Scale the packets/bytes per second to be a total number of
-- of packets/bytes
round(NEW.bytes_per_second * overlap_seconds)::INT,
round(NEW.packets_per_second * overlap_seconds)::INT
from with_overlap_seconds
-- We shouldn't have any 0-second overlaps, but let's just be sure
where overlap_seconds > 0
-- If there is already existing data, then increment the bytes/packets values
on conflict (customer_ip, timestamp, as_number) DO UPDATE SET
bytes = EXCLUDED.bytes + traffic.bytes,
packets = EXCLUDED.packets + traffic.packets
;
end;
$body$;
create or replace function update_aggregated_traffic_hourly() returns trigger
language plpgsql
as
$body$
begin
-- Store aggregated data for different resolutions. For each we also store data
-- without the customer information. This way we can efficiently see traffic data
-- for the whole network
PERFORM update_aggregated_traffic(NEW, 'traffic_perdaytraffic','day', True);
PERFORM update_aggregated_traffic(NEW, 'traffic_perdaytraffic','day', False);
PERFORM update_aggregated_traffic(NEW, 'traffic_perhourtraffic','hour', True);
PERFORM update_aggregated_traffic(NEW, 'traffic_perhourtraffic','hour', False);
PERFORM update_aggregated_traffic(NEW, 'traffic_perminutetraffic','minute', True);
PERFORM update_aggregated_traffic(NEW, 'traffic_perminutetraffic','minute', False);
PERFORM update_aggregated_traffic(NEW, 'traffic_persecondtraffic','second', True);
PERFORM update_aggregated_traffic(NEW, 'traffic_persecondtraffic','second', False);
return NEW;
end;
$body$;
create trigger update_aggregated_traffic_hourly_trigger AFTER INSERT ON traffic_flow
FOR EACH ROW EXECUTE PROCEDURE update_aggregated_traffic_hourly();
person
Adam Charnock
schedule
31.01.2021
select tstzrange('1/25/2021 11:35-8', '1/25//2021 12:45-8', '[]')* tstzrange('1/25/2021 12:00-8', '1/25/2021 13:00-8', '[]'); ["2021-01-25 12:00:00-08","2021-01-25 12:45:00-08"]. Где*— оператор пересечения. - person Adrian Klaver schedule 26.01.2021