Building a Real-Time Dashboard with Distributed PostgreSQL (Citus) – Official Example

Data Model

Immutable HTTP log events are stored in a Citus‑distributed table http_request sharded on site_id so that all rows for a given site reside on the same shard.

-- run on the coordinator
CREATE TABLE http_request (
  site_id INT,
  ingest_time TIMESTAMPTZ DEFAULT now(),
  url TEXT,
  request_country TEXT,
  ip_address TEXT,
  status_code INT,
  response_time_msec INT
);
SELECT create_distributed_table('http_request', 'site_id');

The default UDF shard count is 2‑4 × the number of CPU cores; this many shards eases rebalancing when new workers are added.

On Azure Hyperscale (Citus) streaming replication provides high availability, so citus.shard_replication_factor can remain at the default. In environments without streaming replication it should be set to at least 2 for fault tolerance.

Ingesting Data

A background psql loop inserts random test rows every few hundred milliseconds.

DO $$
  BEGIN LOOP
    INSERT INTO http_request (
      site_id, ingest_time, url, request_country,
      ip_address, status_code, response_time_msec
    ) VALUES (
      trunc(random()*32), clock_timestamp(),
      concat('http://example.com/', md5(random()::text)),
      ('{China,India,USA,Indonesia}'::text[])[ceil(random()*4)],
      concat(trunc(random()*250+2),'.',trunc(random()*250+2),'.',trunc(random()*250+2),'.',trunc(random()*250+2))::inet,
      ('{200,404}'::int[])[ceil(random()*2)],
      5+trunc(random()*150)
    );
    COMMIT;
    PERFORM pg_sleep(random()*0.25);
  END LOOP;
END $$;

A simple dashboard query shows per‑minute request counts, successes, errors and average response time for the last five minutes.

SELECT
  site_id,
  date_trunc('minute', ingest_time) AS minute,
  COUNT(1) AS request_count,
  SUM(CASE WHEN status_code BETWEEN 200 AND 299 THEN 1 ELSE 0 END) AS success_count,
  SUM(CASE WHEN status_code BETWEEN 200 AND 299 THEN 0 ELSE 1 END) AS error_count,
  SUM(response_time_msec)/COUNT(1) AS average_response_time_msec
FROM http_request
WHERE date_trunc('minute', ingest_time) > now() - '5 minutes'::interval
GROUP BY site_id, minute
ORDER BY minute ASC;

Two drawbacks appear: the query must scan every raw row for long‑term trends, and storage grows proportionally with ingest rate.

Summarization

A pre‑aggregated table http_request_1min stores one row per site per minute.

CREATE TABLE http_request_1min (
  site_id INT,
  ingest_time TIMESTAMPTZ, -- minute this row represents
  error_count INT,
  success_count INT,
  request_count INT,
  average_response_time_msec INT,
  CHECK (request_count = error_count + success_count),
  CHECK (ingest_time = date_trunc('minute', ingest_time))
);
SELECT create_distributed_table('http_request_1min', 'site_id');
CREATE INDEX http_request_1min_idx ON http_request_1min (site_id, ingest_time);

Because both tables share the same site_id sharding, Citus colocates matching shards on the same worker, making joins fast (co‑location).

A roll‑up function aggregates raw rows into the minute table using INSERT INTO … SELECT. The function records the last roll‑up time in a single‑row table latest_rollup and is intended to run every minute.

CREATE TABLE latest_rollup (
  minute TIMESTAMPTZ PRIMARY KEY,
  CHECK (minute = date_trunc('minute', minute))
);
INSERT INTO latest_rollup VALUES ('1901-10-10');

CREATE OR REPLACE FUNCTION rollup_http_request() RETURNS void AS $$
DECLARE
  curr_rollup_time TIMESTAMPTZ := date_trunc('minute', now());
  last_rollup_time TIMESTAMPTZ := (SELECT minute FROM latest_rollup);
BEGIN
  INSERT INTO http_request_1min (
    site_id, ingest_time, request_count,
    success_count, error_count, average_response_time_msec
  ) SELECT
    site_id,
    date_trunc('minute', ingest_time),
    COUNT(1) AS request_count,
    SUM(CASE WHEN status_code BETWEEN 200 AND 299 THEN 1 ELSE 0 END) AS success_count,
    SUM(CASE WHEN status_code BETWEEN 200 AND 299 THEN 0 ELSE 1 END) AS error_count,
    SUM(response_time_msec)/COUNT(1) AS average_response_time_msec
  FROM http_request
  WHERE date_trunc('minute', ingest_time) <@ tstzrange(last_rollup_time, curr_rollup_time, '(]')
  GROUP BY site_id, date_trunc('minute', ingest_time);

  UPDATE latest_rollup SET minute = curr_rollup_time;
END;
$$ LANGUAGE plpgsql;

Schedule the function with a crontab entry on the coordinator:

* * * * * psql -c 'SELECT rollup_http_request();'

Or use the pg_cron extension (GitHub repository: https://github.com/citusdata/pg_cron).

After roll‑up, the dashboard query becomes a simple scan of the minute table:

SELECT site_id, ingest_time AS minute, request_count,
       success_count, error_count, average_response_time_msec
FROM http_request_1min
WHERE ingest_time > date_trunc('minute', now()) - '5 minutes'::interval;

Expiring Old Data

Delete raw rows older than one day and minute aggregates older than one month:

DELETE FROM http_request WHERE ingest_time < now() - interval '1 day';
DELETE FROM http_request_1min WHERE ingest_time < now() - interval '1 month';

These deletions can be wrapped in a function and run by the same cron schedule.

Approximate Distinct Count

For queries like “how many unique visitors last month?” the hyperloglog (HLL) extension stores a 1280‑byte sketch with ~2.2 % error.

Install the extension (already present on Hyperscale):

CREATE EXTENSION hll;

Add an hll column to the minute table and populate it in the roll‑up:

ALTER TABLE http_request_1min ADD COLUMN distinct_ip_addresses hll;

INSERT INTO http_request_1min (..., distinct_ip_addresses)
SELECT ..., hll_add_agg(hll_hash_text(ip_address)) AS distinct_ip_addresses
FROM http_request ...;

Dashboard queries retrieve the approximate unique IP count with hll_cardinality and can compute distinct counts over a window using hll_union_agg:

SELECT hll_cardinality(distinct_ip_addresses) AS distinct_ip_address_count
FROM http_request_1min
WHERE ingest_time > date_trunc('minute', now()) - '5 minutes'::interval;

SELECT hll_cardinality(hll_union_agg(distinct_ip_addresses))
FROM http_request_1min
WHERE ingest_time > date_trunc('minute', now()) - '5 minutes'::interval;

Using JSONB for Unstructured Data

To count visitors per country without adding a column per country, add a JSONB column country_counters to the minute table.

ALTER TABLE http_request_1min ADD COLUMN country_counters JSONB;

The roll‑up now aggregates per‑country counts into a JSON object:

INSERT INTO http_request_1min (..., country_counters)
SELECT ..., jsonb_object_agg(request_country, country_count) AS country_counters
FROM (
  SELECT *, count(1) OVER (PARTITION BY site_id, date_trunc('minute', ingest_time), request_country) AS country_count
  FROM http_request
) h;

Dashboard queries can extract a specific country’s count, e.g., for the USA:

SELECT request_count, success_count, error_count, average_response_time_msec,
       COALESCE(country_counters->>'USA','0')::int AS american_visitors
FROM http_request_1min
WHERE ingest_time > date_trunc('minute', now()) - '5 minutes'::interval;

More Resources

https://github.com/citusdata/real-time-analytics-Hands-On-Lab-Hyperscale-Citus

https://github.com/citusdata/postgres-analytics-tutorial

https://docs.citusdata.com/en/v10.2/sharding/data_modeling.html#colocation

https://docs.citusdata.com/en/v10.2/develop/reference_sql.html#count-distinct

https://docs.citusdata.com/en/v10.2/use_cases/timeseries.html#timeseries

https://github.com/citusdata/postgresql-hll