我需要在 Postgresql 中存储 netflow 数据。这是有关网络流量的数据。每条记录包含以下内容：
我的问题是：如何存储这些数据，以便有效地计算过去 X 天/小时的数据传输率？例如，我可能想绘制过去 7 天到 Netflix 的 ASN 的所有流量的图表，每小时分辨率。
连接开始和结束时间之间的差异可能是毫秒，也可能超过一个小时。
我的第一步是将连接存储在带有 GiST 索引的 TSTZRANGE 字段中。然后，查询过去 7 天的每小时流量数据：
然而，这听起来像是很多繁重的工作。谁能想到更好的选择？
初稿：
WITH ts_bucket AS (
SELECT
LAG(gs, 1) OVER () AS begin_period,
gs AS end_period
FROM
generate_series('1/25/2021 0:00-8'::timestamptz, '1/26/2021 0:00-8'::timestamptz, '1 hour') AS gs
),
se AS (
SELECT
1000000 AS bytes,
'01/25/2021 11:35-8'::timestamptz AS start_ts,
'01/25/2021 12:45-08'::timestamptz AS end_ts
)
SELECT
*,
extract('epoch' FROM (upper(tstzrange(begin_period, end_period, '[]') * tstzrange(start_ts, end_ts, '[]'))) - (lower(tstzrange(begin_period, end_period, '[]') * tstzrange(start_ts, end_ts, '[]')))) * bytes / extract('epoch' FROM end_ts - start_ts) AS data_transferred
FROM
ts_bucket,
se
WHERE
begin_period IS NOT NULL
AND tstzrange(se.start_ts, se.end_ts, '[]') && tstzrange(ts_bucket.begin_period, ts_bucket.end_period, '[]');
begin_period | end_period | bytes | start_ts | end_ts | data_transferred
------------------------+------------------------+---------+------------------------+------------------------+--------------------
2021-01-25 11:00:00-08 | 2021-01-25 12:00:00-08 | 1000000 | 2021-01-25 11:35:00-08 | 2021-01-25 12:45:00-08 | 357142.85714285716
2021-01-25 12:00:00-08 | 2021-01-25 13:00:00-08 | 1000000 | 2021-01-25 11:35:00-08 | 2021-01-25 12:45:00-08 | 642857.1428571428
这是基于连接开始和结束时间存储在单独的字段中，然后根据需要将它们转换为范围。
在对此进行了更多研究之后，我认为真正的答案是没有一种开箱即用的方法可以以高性能的方式实现这一目标。尤其是随着数据量的扩大。最终，聚合数千行会很慢，因为这只是大量的数据访问。
相反，我走了一条不同的路线。我在存储原始流 ( traffic_flow) 的表上使用 Postgresql 触发器。每次将记录插入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();