Citus는 대용량 데이터 세트에 대해 실시간 쿼리를 제공합니다. Citus에서 자주 처리하는 작업 중 하나는 이벤트 데이터를 기반으로 하는 실시간 대시보드 지원입니다.

예를 들어, 다른 기업들의 HTTP 트래픽을 모니터링하도록 돕는 클라우드 서비스 제공자가 있다고 가정해 보겠습니다. 고객이 하나의 HTTP 요청을 받을 때마다 해당 서비스는 로그 기록을 전달받습니다. 이러한 모든 로그를 수집하고 이를 바탕으로 HTTP 분석 대시보드를 생성하여 고객에게 웹사이트의 HTTP 오류 수와 같은 통찰을 제공할 수 있습니다. 중요한 점은 최소한의 지연 시간으로 데이터를 표시해야 한다는 것입니다. 또한 대시보드에는 역사적 추세 그래프도 포함되어야 합니다.

또는 광고 네트워크를 운영하며 고객에게 광고 시리즈의 클릭률을 보여주는 경우도 있을 수 있습니다. 이 경우에도 지연 시간이 중요하며 원시 데이터 양이 많고, 역사적 및 실시간 데이터가 모두 필요합니다.

이 섹션에서는 첫 번째 예제의 일부를 어떻게 구현하는지 설명하지만, 동일한 아키텍처는 두 번째 예제와 많은 다른 사용 사례에도 적용됩니다.

• real-time-analytics-Hands-On-Lab-Hyperscale-Citus
• https://github.com/citusdata/real-time-analytics-Hands-On-Lab-Hyperscale-Citus
• 고객을 위한 실시간 분석 설계하기
• https://github.com/citusdata/postgres-analytics-tutorial

데이터 모델

처리하는 데이터는 변경되지 않는 로그 데이터 스트림입니다. 일반적으로 데이터는 Kafka와 같은 도구를 통해 라우팅되며, 이후 Citus에 직접 삽입됩니다. 데이터를 미리 집계하는 것은 데이터 처리량이 관리하기 어려울 정도로 커질 경우 유용합니다.

다음은 HTTP 이벤트 데이터를 삽입하기 위한 간단한 스키마입니다:

CREATE TABLE http_event (
  site_id INT,
  log_time TIMESTAMPTZ DEFAULT now(),

  resource TEXT,
  visitor_country TEXT,
  client_ip TEXT,

  status INT,
  latency_ms INT
);

SELECT create_distributed_table('http_event', 'site_id');

위 코드에서 create_distributed_table 함수는 site_id 열을 기준으로 http_event 테이블을 해싱하여 배포합니다. 이는 특정 사이트의 모든 데이터가 동일한 샤드에 존재하게 만듭니다.

데이터 삽입

데이터를 삽입한 후 아래와 같은 쿼리를 실행하여 대시보드를 구성할 수 있습니다:

DO $$
BEGIN
  LOOP
    INSERT INTO http_event (site_id, log_time, resource, visitor_country, client_ip, status, latency_ms)
    VALUES (
      trunc(random() * 32),
      clock_timestamp(),
      'page-' || md5(random()::text),
      ARRAY['KR', 'US', 'JP', 'CN'][floor(random() * 4 + 1)],
      concat_ws('.', floor(random() * 256), floor(random() * 256), floor(random() * 256), floor(random() * 256)),
      ARRAY[200, 404][floor(random() * 2 + 1)],
      floor(random() * 150 + 5)
    );
    PERFORM pg_sleep(random() * 0.2);
  END LOOP;
END $$;

데이터 조회

다음과 같은 쿼리를 실행하여 대시보드 데이터를 조회할 수 있습니다:

SELECT
  site_id,
  date_trunc('minute', log_time) AS minute,
  COUNT(*) AS total_requests,
  SUM((status BETWEEN 200 AND 299)::int) AS success_count,
  SUM((status NOT BETWEEN 200 AND 299)::int) AS error_count,
  AVG(latency_ms) AS avg_latency
FROM http_event
WHERE log_time > now() - INTERVAL '5 minutes'
GROUP BY site_id, minute
ORDER BY minute ASC;

데이터 집계

위 설정은 효과적이지만 다음과 같은 단점이 있습니다:

• 차트를 생성할 때마다 모든 로그 행을 검색해야 합니다.
• 저장 비용이 데이터 수집 속도와 질의 가능한 역사 길이에 비례하여 증가합니다.

이 문제를 해결하기 위해 원본 데이터를 사전 집계된 형태로 변환할 수 있습니다. 여기서는 1분 단위로 데이터를 요약하는 테이블을 생성하겠습니다:

CREATE TABLE aggregated_http_event (
  site_id INT,
  interval_start TIMESTAMPTZ,
  
  request_total INT,
  success_total INT,
  error_total INT,
  avg_latency INT,
  
  CHECK (request_total = success_total + error_total),
  CHECK (interval_start = date_trunc('minute', interval_start))
);

SELECT create_distributed_table('aggregated_http_event', 'site_id');

CREATE INDEX agg_http_event_idx ON aggregated_http_event (site_id, interval_start);

이 테이블은 원본 테이블과 동일한 방식으로 site_id를 기준으로 샤딩됩니다. 이로 인해 두 테이블의 샤드는 동일한 워커 노드에 배치됩니다.

데이터를 집계하려면 주기적으로 다음 쿼리를 실행합니다:

CREATE OR REPLACE FUNCTION aggregate_http_events() RETURNS void AS $$
DECLARE
  current_minute TIMESTAMPTZ := date_trunc('minute', now());
  last_aggregated TIMESTAMPTZ := (SELECT MAX(interval_start) FROM aggregated_http_event);
BEGIN
  INSERT INTO aggregated_http_event (
    site_id, interval_start, request_total, success_total, error_total, avg_latency
  )
  SELECT
    site_id,
    date_trunc('minute', log_time),
    COUNT(*),
    SUM((status BETWEEN 200 AND 299)::int),
    SUM((status NOT BETWEEN 200 AND 299)::int),
    AVG(latency_ms)
  FROM http_event
  WHERE log_time >= last_aggregated AND log_time < current_minute
  GROUP BY site_id, date_trunc('minute', log_time);

  RETURN;
END;
$$ LANGUAGE plpgsql;

이 함수는 매 분마다 실행되어야 하며, 이를 위해 cron 작업을 설정할 수 있습니다.

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

새로운 대시보드 쿼리는 다음과 같습니다:

SELECT
  site_id,
  interval_start AS minute,
  request_total,
  success_total,
  error_total,
  avg_latency
FROM aggregated_http_event
WHERE interval_start > date_trunc('minute', now()) - INTERVAL '5 minutes';

데이터 만료

집계를 통해 쿼리 속도를 개선했지만, 무한히 데이터를 유지하면 저장 비용이 증가합니다. 따라서 데이터를 일정 기간 이후로 만료시키는 것이 좋습니다:

DELETE FROM http_event WHERE log_time < now() - INTERVAL '1 day';
DELETE FROM aggregated_http_event WHERE interval_start < now() - INTERVAL '1 month';

근사 고유 카운트

hyperloglog(HLL) 확장은 고유 방문자 수와 같은 근사 카운트를 계산하는 데 유용합니다. 이를 설치하고 사용하면 정확한 카운트보다 더 효율적으로 데이터를 관리할 수 있습니다.

먼저 확장을 설치합니다:

CREATE EXTENSION hll;

그런 다음 집계 테이블에 HLL 열을 추가합니다:

ALTER TABLE aggregated_http_event ADD COLUMN unique_ips hll;

데이터를 집계할 때 다음을 추가합니다:

INSERT INTO aggregated_http_event (
  site_id, interval_start, request_total, success_total, error_total, avg_latency, unique_ips
)
SELECT
  site_id,
  date_trunc('minute', log_time),
  COUNT(*),
  SUM((status BETWEEN 200 AND 299)::int),
  SUM((status NOT BETWEEN 200 AND 299)::int),
  AVG(latency_ms),
  hll_add_agg(hll_hash_text(client_ip))
FROM http_event
GROUP BY site_id, date_trunc('minute', log_time);

대시보드 쿼리에서도 이를 반영할 수 있습니다:

SELECT
  site_id,
  interval_start,
  request_total,
  success_total,
  error_total,
  avg_latency,
  hll_cardinality(unique_ips) AS distinct_visitors
FROM aggregated_http_event
WHERE interval_start > date_trunc('minute', now()) - INTERVAL '5 minutes';