Citus는 PostgreSQL을 확장하여 분산 처리를 가능하게 하는 익스텐션입니다. 표준 PostgreSQL SELECT 쿼리를 Citus 코디네이터에서 실행하면, Citus가 복잡한 SELECT, GROUP BY, ORDER BY, JOIN 등을 병렬화하여 쿼리 성능을 향상시킵니다. Citus는 SELECT 쿼리를 여러 작은 조각으로 나누고, 각 조각을 워커 노드에 할당하며, 실행을 감독하고, 결과를 병합하여 최종 결과를 사용자에게 반환합니다.

집계 함수(Aggregate Functions)

Citus는 PostgreSQL에서 지원하는 대부분의 집계 함수(사용자 정의 집계 포함)를 지원하고 병렬화합니다. 집계는 다음 세 가지 방법 중 하나로 실행되며, 우선순위는 다음과 같습니다.

1. 분포 키 기준 그룹화: 집계가 테이블의 분포 열을 기준으로 그룹화되면, Citus는 전체 쿼리 실행을 각 워커로 하위 위임합니다. 이 경우 모든 집계가 지원되며 워커에서 병렬로 실행됩니다. (사용자 정의 집계를 사용하는 경우 워커에도 설치되어 있어야 합니다.)
2. 부분 집계(Partial Aggregation): 집계가 분포 열을 기준으로 그룹화되지 않은 경우에도, Citus는 특정 집계 함수에 대해 최적화를 수행할 수 있습니다. 예를 들어, avg()의 경우 Citus는 각 워커에서 합계와 개수를 가져와 코디네이터에서 최종 평균을 계산합니다. 다음은 이러한 특별 처리가 가능한 집계 함수 목록입니다.

avg, min, max, sum, count, array_agg, jsonb_agg, jsonb_object_agg, json_agg, json_object_agg, bit_and, bit_or, bool_and, bool_or, every, hll_add_agg, hll_union_agg, topn_add_agg, topn_union_agg, any_value, var_pop(float4), var_pop(float8), var_samp(float4), var_samp(float8), variance(float4), variance(float8) stddev_pop(float4), stddev_pop(float8), stddev_samp(float4), stddev_samp(float8) stddev(float4), stddev(float8) tdigest(double precision, int), tdigest_percentile(double precision, int, double precision), tdigest_percentile(double precision, int, double precision[]), tdigest_percentile(tdigest, double precision), tdigest_percentile(tdigest, double precision[]), tdigest_percentile_of(double precision, int, double precision), tdigest_percentile_of(double precision, int, double precision[]), tdigest_percentile_of(tdigest, double precision), tdigest_percentile_of(tdigest, double precision[])

3. 코디네이터 집계(Fallback): 위 두 경우에 해당하지 않으면, Citus는 모든 행을 워커에서 가져와 코디네이터에서 집계를 수행합니다. 이는 네트워크 오버헤드를 유발하며, 데이터 세트가 너무 크면 코디네이터의 리소스를 고갈시킬 수 있습니다. 이 방식은 비활성화할 수 있습니다 (아래 참조).

쿼리의 사소한 변경이 실행 방식을 바꿔 효율성에 영향을 줄 수 있습니다. 예를 들어, 비분포 열로 그룹화된 sum(x)는 분산 실행이 가능하지만, sum(distinct x)는 전체 입력 레코드 세트를 코디네이터로 가져와야 합니다.

SELECT sum(value1), sum(distinct value2) FROM distributed_table;

데이터가 코디네이터로 의도치 않게 이동하는 것을 방지하려면 GUC 설정을 사용할 수 있습니다.

SET citus.coordinator_aggregation_strategy TO 'disabled';

참고: 코디네이터 집계 전략을 비활성화하면 '유형 3'(최후의 수단) 집계 쿼리가 완전히 차단됩니다.

Count(Distinct) 집계

Citus는 count(distinct) 집계를 여러 방식으로 지원합니다.

• 분포 열 기준: count(distinct)가 분포 열을 기반으로 하면, Citus는 쿼리를 워커에 직접 하위 위임합니다.
• 비분포 열 기준: 그렇지 않은 경우, Citus는 각 워커에서 select distinct 문을 실행하고 결과 목록을 코디네이터로 반환하여 최종 개수를 계산합니다.

워커에 고유 항목이 많을수록 데이터 전송 속도가 느려집니다. 특히 여러 count(distinct) 집계가 포함된 쿼리의 경우 더욱 그렇습니다.

-- 여러 distinct count가 포함된 쿼리는 느려질 수 있음
SELECT count(distinct a), count(distinct b), count(distinct c)
FROM table_abc;

성능 향상을 위해 근사치 계산을 선택할 수 있습니다. 다음 단계를 따르세요.

1. 모든 PostgreSQL 인스턴스(코디네이터 및 모든 워커)에 hll 확장을 다운로드하여 설치합니다. (GitHub 저장소: https://github.com/citusdata/postgresql-hll)
2. 코디네이터에서 다음 명령을 실행하여 모든 인스턴스에 hll 확장을 생성합니다.

CREATE EXTENSION hll;

3. citus.count_distinct_error_rate 설정 값을 통해 근사 count distinct를 활성화합니다. 값이 낮을수록 더 정확하지만 계산 시간이 더 오래 걸립니다. 0.005를 권장합니다.

SET citus.count_distinct_error_rate to 0.005;

이후 count(distinct) 집계는 자동으로 HLL을 사용합니다.

HyperLogLog 컬럼 사용

데이터가 HLL 컬럼으로 저장된 경우, hll_union_agg(hll_column)을 호출하여 동적으로 집계할 수 있습니다.

Top N 항목 추정

데이터 세트가 커질수록 count, sort, limit을 사용한 정확한 Top N 계산은 느리고 리소스를 많이 소모합니다. TopN 확장을 사용하면 근사 결과를 빠르게 얻을 수 있습니다.

참고: https://github.com/citusdata/postgresql-topn

기본 연산

select topn_add('{}', 'a');
-- => {"a": 1}

select topn_add(topn_add('{}', 'a'), 'a');
-- => {"a": 2}

-- 정규 분포에서 값 계산
SELECT topn_add_agg(floor(abs(i))::text)
  FROM normal_rand(1000, 5, 0.7) i;
-- => {"2": 1, "3": 74, "4": 420, "5": 425, "6": 77, "7": 3}

실제 예제

2000년 아마존 제품 리뷰 데이터 세트를 사용합니다.

curl -L https://examples.citusdata.com/customer_reviews_2000.csv.gz | \
  gunzip > reviews.csv

CREATE TABLE customer_reviews
(
    customer_id TEXT,
    review_date DATE,
    review_rating INTEGER,
    review_votes INTEGER,
    review_helpful_votes INTEGER,
    product_id CHAR(10),
    product_title TEXT,
    product_sales_rank BIGINT,
    product_group TEXT,
    product_category TEXT,
    product_subcategory TEXT,
    similar_product_ids CHAR(10)[]
);

SELECT create_distributed_table('customer_reviews', 'product_id');

\COPY customer_reviews FROM 'reviews.csv' WITH CSV

CREATE EXTENSION topn;

CREATE TABLE reviews_by_day
(
  review_date date unique,
  agg_data jsonb
);

SELECT create_reference_table('reviews_by_day');

INSERT INTO reviews_by_day
  SELECT review_date, topn_add_agg(product_id)
  FROM customer_reviews
  GROUP BY review_date;

-- 첫 5일 동안 각 날짜별로 가장 많이 리뷰된 제품 조회
SELECT review_date, (topn(agg_data, 1)).*
FROM reviews_by_day
ORDER BY review_date
LIMIT 5;

 review_date |    item    | frequency
-------------+------------+-----------
 2000-01-01  | 0939173344 |        12
 2000-01-02  | B000050XY8 |        11
 2000-01-03  | 0375404368 |        12
 2000-01-04  | 0375408738 |        14
 2000-01-05  | B00000J7J4 |        17

-- 첫 달 전체 데이터 병합 후 상위 5개 제품 조회
SELECT (topn(topn_union_agg(agg_data), 5)).*
FROM reviews_by_day
WHERE review_date >= '2000-01-01' AND review_date < '2000-02-01'
ORDER BY 2 DESC;

    item    | frequency
------------+-----------
 0375404368 |       217
 0345417623 |       217
 0375404376 |       217
 0375408738 |       217
 043936213X |       204

백분위수 계산

정확한 백분위수 계산은 모든 행을 코디네이터로 전송해야 하므로 비용이 많이 듭니다. t-digest 확장을 사용하면 워커에서 병렬로 근사치를 계산할 수 있습니다.

1. 모든 PostgreSQL 노드에 tdigest 확장을 설치합니다. (GitHub 저장소: https://github.com/tvondra/tdigest)
2. 코디네이터에서 확장을 생성합니다.

CREATE EXTENSION tdigest;

Citus는 쿼리에서 tdigest 집계를 사용할 때 부분 계산을 워커로 하위 위임합니다. 정밀도는 compression 매개변수로 제어할 수 있습니다.

LIMIT 하위 위임 (Pushdown)

Citus는 가능한 경우 LIMIT 절을 워커의 샤드로 하위 위임하여 네트워크 전송 데이터를 최소화합니다.

그러나 LIMIT이 있는 SELECT 쿼리가 정확한 결과를 생성하기 위해 모든 샤드에서 모든 행을 가져와야 하는 경우도 있습니다. 예를 들어, 집계 열 기준 정렬이 필요한 경우입니다. 이 경우 근사치가 유용할 수 있으며, citus.limit_clause_row_fetch_count 설정을 통해 활성화할 수 있습니다.

SET citus.limit_clause_row_fetch_count to 10000;

분산 테이블 뷰 (Views)

Citus는 분산 테이블에 대한 모든 뷰를 지원합니다. PostgreSQL CREATE VIEW 문서를 참조하세요.

일부 뷰는 쿼리 계획의 효율성을 저하시킬 수 있습니다. 성능 튜닝에 대한 자세한 내용은 서브쿼리/CTE 네트워크 오버헤드 문서를 참조하세요. Citus는 구체화된 뷰(Materialized View)도 지원하며, 이를 코디네이터 노드의 로컬 테이블로 저장합니다.

조인 (JOIN)

Citus는 테이블 크기와 분산 방식에 관계없이 테이블 간의 동등 조인(equi-JOIN)을 지원합니다. 쿼리 플래너는 최적의 조인 방법과 순서를 선택합니다.

• 공동 배치 조인 (Co-located Join): 두 테이블이 공동 배치된 경우, 공통 분포 열을 기준으로 효율적으로 조인할 수 있습니다. 대규모 분산 테이블 조인에 가장 효율적입니다. (동일한 수의 샤드와 정확히 일치하는 분포 열 타입이 필요합니다.)
• 참조 테이블 조인 (Reference Table Join): 참조 테이블은 모든 워커에 복제되므로, 각 워커에서 로컬 조인으로 분해되어 병렬 실행됩니다. 공동 배치 조인보다 더 유연하며, 테이블의 모든 열과 자유롭게 조인할 수 있습니다.
• 재분할 조인 (Repartition Join): 분포 열 이외의 열을 기준으로 두 테이블을 조인해야 할 때 사용됩니다. 쿼리 최적화기가 분포 열, 조인 키, 테이블 크기에 따라 재분할할 테이블을 결정하며, 관련 샤드만 서로 연결되도록 하여 네트워크 전송 데이터를 줄입니다.

일반적으로 공동 배치 조인이 재분할 조인보다 효율적입니다. 가능하면 공통 조인 키로 테이블을 분산하는 것이 좋습니다.