ClickHouse에서 쿼리를 병렬로 실행하는 방식

ClickHouse는 속도를 위해 설계되었습니다. 쿼리를 매우 높은 수준의 병렬 방식으로 실행하며, 사용 가능한 모든 CPU 코어를 활용하고, 데이터를 처리 레인에 분산하고, 하드웨어 성능을 한계에 가깝게 끌어올리는 경우가 많습니다. 이 가이드에서는 ClickHouse에서 쿼리 병렬성이 어떻게 작동하는지, 그리고 대규모 워크로드에서 성능을 개선하기 위해 이를 어떻게 조정하거나 모니터링할 수 있는지 설명합니다. 핵심 개념을 설명하기 위해 uk_price_paid_simple 데이터셋에 대한 집계 쿼리를 사용합니다.

단계별: ClickHouse가 집계 쿼리를 병렬화하는 방법

ClickHouse는 ① 테이블의 프라이머리 키(primary key)에 대한 필터가 있는 집계 쿼리를 실행할 때, ② 어떤 그래뉼을 처리해야 하고 어떤 그래뉼을 안전하게 건너뛸 수 있는지 식별하기 위해 ③ 프라이머리 인덱스(primary index)를 메모리에 로드합니다:

처리 레인 간 작업 분산

선택한 데이터는 이후 n개의 병렬 처리 레인에 동적으로 분산되며, 각 레인은 데이터를 블록 단위로 스트리밍하고 처리해 최종 결과를 생성합니다:

n개의 병렬 처리 레인 수는 max_threads 설정으로 제어되며, 기본적으로 서버에서 ClickHouse가 사용할 수 있는 단일 CPU의 코어 수(스레드 수)와 일치합니다. 위 예시에서는 4개의 코어를 가정합니다. 8개의 코어가 있는 머신에서는 더 많은 레인이 데이터를 병렬로 처리하므로 쿼리 처리량이 대략 2배로 증가합니다(메모리 사용량도 그에 따라 함께 증가합니다):

효율적인 레인 분산은 CPU 활용도를 극대화하고 전체 쿼리 시간을 줄이는 핵심입니다.

세그먼트된 테이블에서 쿼리 처리하기

테이블 데이터가 여러 서버에 세그먼트로 분산되어 있으면 각 서버는 자신의 세그먼트를 병렬로 처리합니다. 각 서버 내부에서는 앞서 설명한 대로 로컬 데이터가 병렬 처리 레인을 통해 처리됩니다:

처음 쿼리를 받은 서버는 세그먼트에서 나온 모든 하위 결과를 수집해 최종 전역 결과로 결합합니다. 쿼리 부하를 세그먼트 전체에 분산하면 병렬성을 수평으로 확장할 수 있으며, 특히 처리량이 높은 환경에서 효과적입니다.

ClickHouse Cloud는 세그먼트 대신 병렬 레플리카를 사용합니다ClickHouse Cloud에서는 동일한 병렬성을 병렬 레플리카를 통해 구현하며, 이는 shared-nothing 클러스터의 세그먼트와 유사하게 동작합니다. 각 ClickHouse Cloud 레플리카(무상태 컴퓨트 노드)는 데이터의 일부를 병렬로 처리하고, 독립적인 세그먼트와 마찬가지로 최종 결과 생성에 기여합니다.

쿼리 병렬성 모니터링

다음 도구를 사용해 쿼리가 사용 가능한 CPU 리소스를 충분히 활용하는지 확인하고, 그렇지 않을 때 원인을 진단할 수 있습니다. 이 예시는 59개의 CPU 코어가 있는 테스트 서버에서 실행하며, 이를 통해 ClickHouse의 쿼리 병렬성을 충분히 보여줄 수 있습니다. 예시 쿼리가 어떻게 실행되는지 확인하기 위해, 집계 쿼리 수행 중 trace 수준의 모든 로그 엔트리를 반환하도록 ClickHouse 서버에 지시할 수 있습니다. 이번 시연에서는 쿼리의 프레디케이트를 제거했습니다. 그렇지 않으면 3개의 그래뉼만 처리되어, ClickHouse가 몇 개 이상의 병렬 처리 레인을 활용하기에는 데이터가 충분하지 않기 때문입니다:

SELECT
   max(price)
FROM
   uk.uk_price_paid_simple
SETTINGS send_logs_level='trace';

① <Debug> ...: 3609 marks to read from 3 ranges
② <Trace> ...: Spreading mark ranges among streams
② <Debug> ...: Reading approx. 29564928 rows with 59 streams

다음과 같은 점을 확인할 수 있습니다.

① ClickHouse는 3개의 데이터 범위에 걸쳐 3,609개의 그래뉼(트레이스 로그에서는 마크로 표시됨)을 읽어야 합니다.
② 59개의 CPU 코어가 있으므로 이 작업은 59개의 병렬 처리 스트림에 분산되며, 각 레인에 하나씩 할당됩니다.

또는 EXPLAIN 절을 사용해 집계 쿼리의 물리적 연산자 계획, 즉 “쿼리 파이프라인”이라고도 하는 구조를 살펴볼 수 있습니다.

EXPLAIN PIPELINE
SELECT
   max(price)
FROM
   uk.uk_price_paid_simple;

    ┌─explain───────────────────────────────────────────────────────────────────────────┐
│ (Expression)                                                                      │
│ ExpressionTransform × 59                                                          │
│   (Aggregating)                                                                   │
│   Resize 59 → 59                                                                  │
│     AggregatingTransform × 59                                                     │
│       StrictResize 59 → 59                                                        │
│         (Expression)                                                              │
│         ExpressionTransform × 59                                                  │
│           (ReadFromMergeTree)                                                     │
│           MergeTreeSelect(pool: PrefetchedReadPool, algorithm: Thread) × 59 0 → 1 │
    └───────────────────────────────────────────────────────────────────────────────────┘

참고: 위의 연산자 계획은 아래에서 위로 읽으십시오. 각 줄은 물리적 실행 계획의 한 단계를 나타내며, 맨 아래의 스토리지에서 데이터를 읽는 단계부터 시작해 맨 위의 최종 처리 단계로 끝납니다. × 59로 표시된 연산자는 59개의 병렬 처리 레인에서 서로 겹치지 않는 데이터 영역에 대해 동시에 실행됩니다. 이는 max_threads 값을 반영하며, 쿼리의 각 단계가 CPU 코어 전반에서 어떻게 병렬화되는지 보여줍니다. ClickHouse의 내장 web UI (/play endpoint에서 사용 가능)는 위의 물리적 계획을 그래픽으로 시각화해 표시할 수 있습니다. 이 예시에서는 시각화를 간결하게 유지하기 위해 max_threads를 4로 설정하여 병렬 처리 레인 4개만 표시합니다: 참고: 이 시각화는 왼쪽에서 오른쪽으로 읽으십시오. 각 행은 데이터 블록을 순차적으로 스트리밍하면서 필터링, 집계, 최종 처리 단계와 같은 변환을 적용하는 병렬 처리 레인을 나타냅니다. 이 예시에서는 max_threads = 4 설정에 해당하는 4개의 병렬 레인을 확인할 수 있습니다.

처리 레인 간 부하 분산

위의 물리 계획에서 Resize 연산자는 데이터 블록 스트림을 재파티셔닝하고 재분배하여 처리 레인이 고르게 활용되도록 합니다. 이러한 재균형은 데이터 범위별로 쿼리 프레디케이트와 일치하는 행 수가 다를 때 특히 중요합니다. 그렇지 않으면 일부 레인에는 부하가 집중되고 다른 레인은 유휴 상태로 남을 수 있습니다. 작업을 재분배하면 더 빠른 레인이 더 느린 레인의 작업을 사실상 나눠 처리하게 되어 전체 쿼리 런타임이 최적화됩니다.

max_threads가 항상 준수되지는 않는 이유

위에서 언급했듯이, n개의 병렬 처리 레인 수는 max_threads 설정으로 제어되며, 기본적으로 서버에서 ClickHouse가 사용할 수 있는 CPU 코어 수와 같습니다:

SELECT getSetting('max_threads');

   ┌─getSetting('max_threads')─┐
1. │                        59 │
   └───────────────────────────┘

하지만 처리할 데이터로 선택된 양에 따라 max_threads 값이 적용되지 않을 수 있습니다:

EXPLAIN PIPELINE
SELECT
   max(price)
FROM
   uk.uk_price_paid_simple
WHERE town = 'LONDON';

...   
(ReadFromMergeTree)
MergeTreeSelect(pool: PrefetchedReadPool, algorithm: Thread) × 30

위의 연산자 계획 추출 내용에서 볼 수 있듯이, max_threads를 59로 설정했더라도 ClickHouse는 데이터를 스캔할 때 동시 스트림을 30개만 사용합니다. 이제 쿼리를 실행해 보겠습니다:

SELECT
   max(price)
FROM
   uk.uk_price_paid_simple
WHERE town = 'LONDON';

   ┌─max(price)─┐
1. │  594300000 │ -- 594.30 million
   └────────────┘
   
1 row in set. Elapsed: 0.013 sec. Processed 2.31 million rows, 13.66 MB (173.12 million rows/s., 1.02 GB/s.)
Peak memory usage: 27.24 MiB.   

위 출력에서 볼 수 있듯이, 이 쿼리는 231만 개의 행을 처리하고 13.66MB의 데이터를 읽었습니다. 이는 인덱스 분석 단계에서 ClickHouse가 처리 대상으로 282개의 그래뉼을 선택했기 때문이며, 각 그래뉼에는 8,192개의 행이 포함되어 있어 총 약 231만 개의 행이 됩니다:

EXPLAIN indexes = 1
SELECT
   max(price)
FROM
   uk.uk_price_paid_simple
WHERE town = 'LONDON';

    ┌─explain───────────────────────────────────────────────┐
│ Expression ((Project names + Projection))             │
│   Aggregating                                         │
│     Expression (Before GROUP BY)                      │
│       Expression                                      │
│         ReadFromMergeTree (uk.uk_price_paid_simple)   │
│         Indexes:                                      │
│           PrimaryKey                                  │
│             Keys:                                     │
│               town                                    │
│             Condition: (town in ['LONDON', 'LONDON']) │
│             Parts: 3/3                                │
│             Granules: 282/3609                        │
    └───────────────────────────────────────────────────────┘  

설정된 max_threads 값과 관계없이, ClickHouse는 이를 뒷받침할 만큼 데이터가 충분할 때에만 추가 처리 레인을 할당합니다. max_threads의 “max”는 실제 사용되는 스레드 수를 보장한다는 뜻이 아니라 상한을 의미합니다. 여기서 “충분한 데이터”는 주로 두 가지 설정으로 결정되며, 각 처리 레인이 처리해야 하는 최소 행 수(기본값 163,840)와 최소 바이트 수(기본값 2,097,152)를 정의합니다. shared-nothing 클러스터의 경우:

공유 스토리지를 사용하는 클러스터의 경우(예: ClickHouse Cloud):

또한 읽기 작업 크기에는 절대적인 하한이 있으며, 다음 설정으로 제어됩니다.

Merge_tree_min_read_task_size + merge_tree_min_bytes_per_task_for_remote_reading

이 설정은 수정하지 마십시오운영 환경에서는 이 설정을 수정하지 않는 것이 좋습니다. 여기서는 max_threads가 실제 병렬성 수준을 항상 결정하지는 않는 이유를 설명하기 위해서만 보여줍니다.

시연을 위해, 최대 동시성이 강제로 적용되도록 이 설정을 재정의한 상태에서 물리 계획을 살펴보겠습니다.

EXPLAIN PIPELINE
SELECT
   max(price)
FROM
   uk.uk_price_paid_simple
WHERE town = 'LONDON'
SETTINGS
  max_threads = 59,
  merge_tree_min_read_task_size = 0,
  merge_tree_min_rows_for_concurrent_read_for_remote_filesystem = 0, 
  merge_tree_min_bytes_for_concurrent_read_for_remote_filesystem = 0;

...   
(ReadFromMergeTree)
MergeTreeSelect(pool: PrefetchedReadPool, algorithm: Thread) × 59

이제 ClickHouse는 설정된 max_threads를 완전히 준수하면서 59개의 동시 스트림을 사용해 데이터를 스캔합니다. 이는 작은 데이터셋에 대한 쿼리에서는 ClickHouse가 의도적으로 동시성을 제한한다는 점을 보여줍니다. 설정 재정의는 비효율적인 실행이나 리소스 경합을 초래할 수 있으므로 테스트 용도로만 사용하고, 프로덕션 환경에서는 사용하지 마십시오.

핵심 요약

ClickHouse는 max_threads에 연결된 처리 레인을 사용해 쿼리를 병렬로 실행합니다.
실제 레인 수는 처리할 데이터로 선택된 데이터의 크기에 따라 달라집니다.
레인 사용 방식을 분석하려면 EXPLAIN PIPELINE과 트레이스 로그를 사용하세요.

더 많은 정보를 확인할 수 있는 곳

ClickHouse가 쿼리를 병렬로 실행하는 방식과 대규모 환경에서 높은 성능을 달성하는 방법을 더 자세히 알아보려면, 다음 자료를 참고하십시오:

쿼리 처리 레이어 – VLDB 2024 논문(웹 버전) - 스케줄링, 파이프라이닝, 연산자 설계를 포함해 ClickHouse의 내부 실행 모델을 자세히 설명합니다.
부분 집계 상태 설명 - 부분 집계 상태가 처리 레인 전반에서 효율적인 병렬 실행을 어떻게 가능하게 하는지 기술적으로 자세히 설명합니다.
ClickHouse 쿼리 처리의 전체 단계를 자세히 설명하는 비디오 튜토리얼:

​단계별: ClickHouse가 집계 쿼리를 병렬화하는 방법

​처리 레인 간 작업 분산

​세그먼트된 테이블에서 쿼리 처리하기

​쿼리 병렬성 모니터링

​처리 레인 간 부하 분산

​max_threads가 항상 준수되지는 않는 이유

​핵심 요약

​더 많은 정보를 확인할 수 있는 곳

단계별: ClickHouse가 집계 쿼리를 병렬화하는 방법

처리 레인 간 작업 분산

세그먼트된 테이블에서 쿼리 처리하기

쿼리 병렬성 모니터링

처리 레인 간 부하 분산

max_threads가 항상 준수되지는 않는 이유

핵심 요약

더 많은 정보를 확인할 수 있는 곳