ClickHouse가 쿼리를 병렬로 실행하는 방법

ClickHouse는 사용 가능한 모든 CPU 코어를 활용하고 데이터를 프로세싱 레인(processing lanes)에 분산하여 쿼리를 고도로 병렬로 실행합니다. 이 문서에서는 쿼리 병렬 처리가 어떻게 동작하는지, 그리고 max_threads 설정으로 그것을 어떻게 튜닝하고 모니터링하는지 단계별로 설명할게요.

출처: 문서

본문

ClickHouse는 속도를 위해 설계되었습니다. 사용 가능한 모든 CPU 코어를 활용하고, 데이터를 프로세싱 레인에 분산하며, 종종 하드웨어를 한계까지 밀어 붙이면서 쿼리를 고도로 병렬로 실행합니다. 이 가이드는 ClickHouse에서 쿼리 병렬 처리가 어떻게 동작하는지, 그리고 대규모 워크로드에서 성능을 개선하기 위해 그것을 어떻게 튜닝하거나 모니터링하는지 살펴봅니다. 핵심 개념을 설명하기 위해 uk_price_paid_simple 데이터셋에 대한 집계 쿼리를 사용합니다.

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

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

프로세싱 레인에 작업 분산하기

선택된 데이터는 그다음 n개의 병렬 프로세싱 레인에 동적으로 분산되며, 이 레인들은 데이터를 블록 단위로 스트리밍하고 처리하여 최종 결과로 만듭니다.

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

효율적인 레인 분배는 CPU 활용을 극대화하고 총 쿼리 시간을 줄이는 핵심입니다.

샤딩된 테이블에서 쿼리 처리하기

테이블 데이터가 샤드(shards)로 여러 서버에 분산되어 있으면 각 서버는 자신의 샤드를 병렬로 처리합니다. 각 서버 내에서 로컬 데이터는 위에서 설명한 것처럼 병렬 프로세싱 레인을 사용해 처리됩니다. 처음에 쿼리를 받은 서버는 모든 샤드로부터 하위 결과를 수집하여 최종 전역 결과로 결합합니다. 쿼리 부하를 샤드에 분산하면 특히 고처리량 환경에서 병렬 처리의 수평 확장이 가능합니다.

ClickHouse Cloud는 샤드 대신 병렬 복제본(parallel replicas)을 사용합니다. ClickHouse Cloud에서도 동일한 병렬 처리가 병렬 복제본을 통해 달성되는데, 이것은 shared-nothing 클러스터의 샤드와 유사하게 동작합니다. 각 ClickHouse Cloud 복제본 — 상태 비저장(stateless) 컴퓨팅 노드 — 은 데이터의 일부를 병렬로 처리하고 최종 결과에 기여하며, 마치 독립적인 샤드처럼 동작합니다.

쿼리 병렬 처리 모니터링하기

다음 도구를 사용해 쿼리가 사용 가능한 CPU 리소스를 완전히 활용하는지 확인하고, 그렇지 않을 때 진단하세요. 여기서는 ClickHouse가 쿼리 병렬 처리를 완전히 보여줄 수 있도록 59개 CPU 코어를 가진 테스트 서버에서 실행합니다. 예제 쿼리가 어떻게 실행되는지 관찰하기 위해 ClickHouse 서버에 집계 쿼리 중 모든 trace 레벨 로그 항목을 반환하도록 지시할 수 있습니다. 이 데모에서는 쿼리의 조건을 제거했습니다. 그렇지 않으면 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개 그래뉼(trace 로그에서 마크로 표시)을 읽어야 합니다.
② 59개 CPU 코어로 이 작업을 59개 병렬 프로세싱 스트림 — 레인당 하나 — 에 분산합니다.

또는 집계 쿼리의 물리적 연산자 계획(일명 "쿼리 파이프라인")을 검사하기 위해 EXPLAIN 절을 사용할 수 있습니다:

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

    ┌─explain───────────────────────────────────────────────────────────────────────────┐
 1. │ (Expression)                                                                      │
 2. │ ExpressionTransform × 59                                                          │
 3. │   (Aggregating)                                                                   │
 4. │   Resize 59 → 59                                                                  │
 5. │     AggregatingTransform × 59                                                     │
 6. │       StrictResize 59 → 59                                                        │
 7. │         (Expression)                                                              │
 8. │         ExpressionTransform × 59                                                  │
 9. │           (ReadFromMergeTree)                                                     │
10. │           MergeTreeSelect(pool: PrefetchedReadPool, algorithm: Thread) × 59 0 → 1 │
    └───────────────────────────────────────────────────────────────────────────────────┘

참고: 위 연산자 계획을 아래에서 위로 읽으세요. 각 줄은 물리적 실행 계획의 한 단계를 나타내며, 아래쪽의 스토리지에서 데이터를 읽는 것부터 시작하여 위쪽의 최종 처리 단계로 끝납니다. × 59로 표시된 연산자는 59개 병렬 프로세싱 레인에 걸쳐 겹치지 않는 데이터 영역에서 동시에 실행됩니다. 이는 max_threads의 값을 반영하며 쿼리의 각 단계가 어떻게 CPU 코어에 걸쳐 병렬화되는지 보여줍니다. ClickHouse의 내장 웹 UI(/play 엔드포인트에서 제공)는 위 물리적 계획을 그래픽 시각화로 렌더링할 수 있습니다. 이 예제에서는 시각화를 컴팩트하게 유지하기 위해 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가 각각 8,192행을 포함하는 282개 그래뉼 을 처리용으로 선택했기 때문이며, 총 약 231만 행입니다.

설정된 max_threads 값과 관계없이 ClickHouse는 데이터가 충분히 있을 때만 추가 병렬 프로세싱 레인을 할당합니다. max_threads의 "max"는 보장된 스레드 수가 아니라 상한을 의미합니다. "충분한 데이터(what enough data means)"는 각 프로세싱 레인이 처리해야 할 최소 행 수(기본 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가 왜 항상 실제 병렬 처리 수준을 결정하지 못하는지 설명하기 위한 목적으로만 보여줍니다.

데모 목적으로 최대 동시성을 강제하도록 이 설정들을 오버라이드한 물리적 계획을 검사해 보겠습니다:

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

이제 ClickHouse는 데이터를 스캔하는 데 59개 동시 스트림을 사용하여 설정된 max_threads를 완전히 존중합니다. 이는 작은 데이터셋에 대한 쿼리에서 ClickHouse가 의도적으로 동시성을 제한한다는 것을 보여줍니다. 설정 오버라이드는 프로덕션이 아니라 테스트에만 사용하세요. 비효율적인 실행이나 리소스 경합을 유발할 수 있습니다.

핵심 요점

ClickHouse는 max_threads에 연결된 프로세싱 레인을 사용해 쿼리를 병렬화합니다.
실제 레인 수는 처리를 위해 선택된 데이터 크기에 따라 달라집니다.
EXPLAIN PIPELINE과 trace 로그를 사용해 레인 사용을 분석하세요.

더 자세한 정보는 어디서?

ClickHouse가 쿼리를 병렬로 실행하고 대규모에서 고성능을 달성하는 방법을 더 깊이 알아보고 싶다면 다음 리소스를 살펴보세요:

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

ClickHouse가 쿼리를 병렬로 실행하는 방법