분산 ClickHouse의 두 축 - 샤딩과 복제는 서로 독립이다

단일 노드 ClickHouse가 한계에 닿으면 노드를 늘리게 됩니다. 여기서 "클러스터를 구성한다"는 말이 두 가지 서로 다른 작업을 뭉뚱그립니다. 데이터를 나눠 담는 것(샤딩)과 같은 데이터를 여러 벌 두는 것(복제)입니다.

ClickHouse는 이 둘을 완전히 분리된 메커니즘으로 구현합니다. 그리고 그 분리를 모른 채 설정하면 데이터가 조용히 유실되거나, 노드를 추가했는데 아무것도 나아지지 않는 상태가 됩니다. 이 글에서는 두 축이 각각 무엇을 하는지, 어디서 서로 얽히는지, 그리고 나중에 바꾸기 어려운 결정이 무엇인지 정리합니다.

1. 두 축은 정말로 독립이다

문서에 분명히 적혀 있습니다.

Replication works at the level of an individual table, not the entire server. A server can store both replicated and non-replicated tables at the same time.

replication does not depend on sharding. Each shard has its own independent replication.

여기서 두 가지를 읽을 수 있습니다. 복제는 서버 단위가 아니라 테이블 단위이고, 샤딩과 아무 관련이 없습니다. 한 서버에 복제되는 테이블과 안 되는 테이블이 공존할 수 있습니다.

구조를 그리면 이렇습니다.

클러스터 (샤드 2 × 레플리카 2)

 shard 1                       shard 2
 ├── replica A ─┐              ├── replica A ─┐
 │              │ Keeper가     │              │ Keeper가
 └── replica B ─┘ 동기화       └── replica B ─┘ 동기화
      ↑                             ↑
      └─────── Distributed 테이블이 둘을 합쳐서 조회 ───────┘

• 샤딩: 어느 샤드에 행을 넣을지. Distributed 엔진과 sharding_key가 담당.
• 복제: 한 샤드 안에서 사본을 맞추는 것. ReplicatedMergeTree와 ClickHouse Keeper가 담당.

두 축을 각각 설정하지 않으면 절반만 동작합니다. 흔한 실수가 Distributed 테이블만 만들고 로컬 테이블을 MergeTree로 두는 것입니다. 이 상태는 샤딩은 되지만 복제가 없어, 노드 하나가 죽으면 그 샤드의 데이터가 통째로 조회에서 빠집니다.

2. Distributed 엔진은 데이터를 저장하지 않는다

가장 먼저 오해하는 지점입니다.

Distributed engine ... do not store any data of their own, but allow distributed query processing on multiple servers.

Distributed 테이블은 뷰에 가까운 라우터입니다. 실제 데이터는 각 노드의 로컬 테이블에 있습니다. 따라서 테이블을 만들 때 항상 두 개를 만듭니다.

-- 1) 각 노드에 로컬 테이블 (실제 저장)
CREATE TABLE events_local ON CLUSTER my_cluster
(
    event_time  DateTime,
    user_id     UInt64,
    event_type  LowCardinality(String),
    payload     String
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events_local', '{replica}')
PARTITION BY toYYYYMM(event_time)
ORDER BY (user_id, event_time);

-- 2) 조회·삽입용 분산 테이블 (저장하지 않음)
CREATE TABLE events ON CLUSTER my_cluster AS events_local
ENGINE = Distributed(my_cluster, currentDatabase(), events_local, cityHash64(user_id));

ReplicatedMergeTree의 두 인자가 복제의 핵심입니다. 첫 번째는 Keeper 안의 조정 경로, 두 번째는 이 서버가 그 경로에서 갖는 이름입니다. {shard}와 {replica}는 매크로입니다.

The substituted values are taken from the macros section of the configuration file.

노드별 설정에 값을 두면 같은 DDL을 모든 노드에 그대로 실행할 수 있습니다. 그래서 ON CLUSTER가 성립합니다.

<!-- shard 1의 replica A 노드 -->
<clickhouse>
  <macros>
    <shard>01</shard>
    <replica>shard01-replica-a</replica>
  </macros>
</clickhouse>

여기서 규칙이 하나 나옵니다. 같은 샤드의 레플리카끼리는 Keeper 경로가 같아야 하고, 다른 샤드끼리는 달라야 합니다. 매크로를 잘못 넣어 두 샤드가 같은 경로를 쓰면, 서로 다른 데이터를 담아야 할 노드들이 같은 복제 그룹으로 묶여 데이터가 뒤섞입니다. 반대로 같은 샤드의 레플리카가 서로 다른 경로를 쓰면 복제가 아예 일어나지 않습니다. 둘 다 오류 없이 조용히 진행됩니다.

3. 샤딩 키는 되돌리기 어려운 결정이다

Distributed의 네 번째 인자가 샤딩 키입니다. 값이 어떻게 쓰이는지는 단순합니다.

the row will be sent to the shard that corresponds to the half-interval of the remainders

나머지 연산입니다. 그리고 여기서 가장 큰 제약이 나옵니다.

data is not automatically rebalanced when adding new shards

샤드를 추가해도 기존 데이터는 옮겨 가지 않습니다. 나눗셈의 제수가 바뀌었으니 새 데이터는 다른 분포로 들어가는데, 옛 데이터는 그대로 있습니다. 결과적으로 샤드 간 크기가 어긋나고, 그것을 바로잡으려면 수동으로 재적재해야 합니다. 테라바이트 단위에서는 며칠짜리 작업입니다.

문서도 나머지 연산의 한계를 인정합니다.

A simple remainder from the division is a limited solution for sharding and isn't always appropriate

그래서 샤딩 키를 정할 때 봐야 할 것이 셋입니다.

분포의 균일성. cityHash64(user_id)처럼 해시를 씌우는 이유입니다. user_id를 그대로 써도 나머지 연산이라 순차 발급된 ID는 고르게 퍼지지만, ID 값 자체가 고르지 않으면(짝수만 발급되거나 특정 대역에 몰리는 경우 등) 특정 샤드로 치우칩니다. 문서도 이럴 때 intHash64(UserID)처럼 해시로 감싸라고 안내합니다. 반대로 event_type 같은 저카디널리티 컬럼을 키로 쓰면 소수 샤드에 데이터가 집중됩니다.

조회의 지역성. 같은 사용자의 데이터가 한 샤드에 모여 있으면 그 사용자의 집계가 한 샤드에서 끝납니다. 여러 샤드를 도는 조회는 네트워크로 중간 결과를 모아야 하므로 비쌉니다. 특히 조인이 그렇습니다. 조인하는 두 테이블을 같은 키로 샤딩하면 각 샤드에서 로컬 조인이 가능해집니다. 다른 키로 샤딩했다면 한쪽을 전 샤드에 뿌려야 합니다.

나중에 바꿀 수 없다는 점. 위 둘을 저울질할 때 이 사실을 기준으로 삼아야 합니다. 애매하면 조인에 쓰이는 키를 고르는 편이 안전합니다. 분포 불균형은 가중치로 어느 정도 보정할 수 있지만, 조인 지역성은 재적재 없이는 얻을 수 없기 때문입니다.

샤드 간 용량이 다르면 클러스터 설정의 가중치로 비율을 조정할 수 있습니다.

<remote_servers>
  <my_cluster>
    <shard>
      <weight>1</weight>
      <internal_replication>true</internal_replication>
      <replica><host>ch-01-a</host><port>9000</port></replica>
      <replica><host>ch-01-b</host><port>9000</port></replica>
    </shard>
    <shard>
      <weight>2</weight>          <!-- 이 샤드에 두 배로 보낸다 -->
      <internal_replication>true</internal_replication>
      <replica><host>ch-02-a</host><port>9000</port></replica>
      <replica><host>ch-02-b</host><port>9000</port></replica>
    </shard>
  </my_cluster>
</remote_servers>

4. internal_replication을 잘못 두면 데이터가 두 벌 쌓인다

위 설정의 internal_replication은 두 축이 만나는 지점입니다. 문서의 설명은 이렇습니다.

When internal_replication is true, ... selects the first healthy replica and writes data to it.

If false, data is written to all replicas ... this approach is worse than using replicated tables because the consistency of replicas is not checked.

true면 Distributed가 샤드당 한 레플리카에만 쓰고, 나머지는 ReplicatedMergeTree의 복제가 채웁니다. false면 Distributed가 모든 레플리카에 직접 씁니다.

로컬 테이블이 ReplicatedMergeTree인데 internal_replication이 false면 두 경로가 동시에 동작합니다. Distributed가 각 레플리카에 쓰고, 복제도 서로에게 전파합니다. 같은 데이터가 중복 삽입될 수 있고 레플리카 간 정합성이 보장되지 않습니다.

기준은 단순합니다.

로컬 테이블 엔진	internal_replication
ReplicatedMergeTree	true (필수)
MergeTree (복제 없음)	false

ReplicatedMergeTree를 쓰면서 이 값을 안 적으면 기본값에 따라 조용히 틀린 조합이 될 수 있으므로, 명시하는 것이 안전합니다.

5. Distributed 테이블에 INSERT하면 생기는 일

읽기와 달리 쓰기 경로에는 함정이 있습니다.

Data is written in background ... the data block is just written to the local file system. The data is sent to the remote servers in the background as soon as possible.

Distributed 테이블에 INSERT하면 데이터가 먼저 그 노드의 디스크에 임시 파일로 쌓이고, 백그라운드로 각 샤드에 전송됩니다. INSERT는 전송 완료를 기다리지 않고 성공을 반환합니다. 그래서 다음이 성립합니다.

If the server ceased to exist or had a rough restart ... after an INSERT to a Distributed table, the inserted data might be lost.

성공 응답을 받은 데이터가 유실될 수 있습니다. 큐가 디스크에 있으므로 정상 재시작은 견디지만, 디스크가 날아가거나 그 노드를 폐기하면 미전송분이 사라집니다.

선택지는 셋입니다.

(A) 동기 전송으로 바꾼다. distributed_foreground_insert를 켜면 INSERT가 샤드 전송 완료까지 기다립니다. 유실은 막지만 처리량이 떨어지고, 샤드 하나가 느리면 INSERT 전체가 느려집니다.

INSERT INTO events SETTINGS distributed_foreground_insert = 1 VALUES (...);

(B) 로컬 테이블에 직접 INSERT한다. 적재 클라이언트가 샤딩을 직접 계산하거나, 노드를 라운드로빈으로 골라 각 노드의 events_local에 씁니다. 중간 큐가 없으므로 유실 경로가 사라집니다. 대신 샤딩 로직이 애플리케이션으로 올라옵니다. 대량 적재에서는 이 방식이 사실상 표준입니다.

(C) 재전송 가능한 원본을 남긴다. Kafka 같은 로그를 앞에 두고 오프셋을 커밋 기준으로 삼습니다. 유실이 나도 다시 읽으면 됩니다.

읽기 전용으로 Distributed를 쓰고 쓰기는 (B)나 (C)로 처리하는 조합이 무난합니다. Distributed에 편하게 INSERT하는 방식은 소량·저빈도일 때만 적절합니다.

복제 쪽에도 비슷한 선택이 있습니다. 복제는 기본적으로 비동기이자 멀티 마스터입니다. 어느 레플리카에 써도 되고, 쓴 직후 다른 레플리카에는 아직 없을 수 있습니다. 쓰기 확정을 강하게 만들려면 insert_quorum으로 몇 개 레플리카가 받았을 때 성공으로 볼지 정합니다. 지연과 가용성을 내주고 내구성을 얻는 교환입니다.

6. Keeper가 멈추면 무엇이 멈추는가

복제 메타데이터는 ClickHouse Keeper(또는 ZooKeeper)에 있습니다.

ClickHouse uses ClickHouse Keeper for storing replicas meta information.

Keeper가 응답하지 않으면 복제 테이블은 읽기 전용으로 떨어집니다. 조회는 되지만 INSERT가 막힙니다. 즉 Keeper 가용성이 곧 쓰기 가용성입니다. 노드 수가 적다고 Keeper를 한 대만 두면 그 한 대가 클러스터 전체의 단일 장애점이 됩니다. 홀수(보통 3대)로, 그리고 ClickHouse 서버와 다른 장애 도메인에 두는 것이 기본입니다.

Keeper 부하는 파티션과 파트 개수에 비례합니다. 파티션을 잘게 나누면(PARTITION BY toDate(...) 같은) 파트가 폭증해 Keeper가 먼저 힘들어집니다. 파티션 키는 스토리지만이 아니라 Keeper 관점에서도 정해야 합니다. 그래서 월 단위가 흔한 기본값입니다.

7. 직접 확인하는 방법

시스템 테이블로 두 축의 상태를 각각 볼 수 있습니다.

-- 클러스터 구성이 의도대로인가 (샤드 번호, 레플리카 배치)
SELECT cluster, shard_num, shard_weight, replica_num, host_name
FROM system.clusters
WHERE cluster = 'my_cluster'
ORDER BY shard_num, replica_num;

-- 매크로가 노드마다 제대로 들어갔는가
SELECT * FROM system.macros;

-- 복제 상태: 지연과 큐 길이
SELECT database, table, is_readonly, is_session_expired,
       absolute_delay, queue_size, inserts_in_queue, merges_in_queue
FROM system.replicas
WHERE is_readonly OR absolute_delay > 60 OR queue_size > 100;

-- 복제 큐에서 막힌 항목의 사유
SELECT database, table, type, num_tries, last_exception
FROM system.replication_queue
WHERE num_tries > 3
LIMIT 20;

-- Distributed 전송 큐 - 여기에 쌓여 있으면 5절의 유실 위험 구간
SELECT database, table, data_files, data_compressed_bytes, last_exception
FROM system.distribution_queue;

system.replicas의 is_readonly가 참이면 Keeper 연결 문제입니다(6절). absolute_delay가 계속 커지면 복제가 따라가지 못하는 상태입니다. system.distribution_queue의 data_files가 0이 아닌 채로 유지되면 샤드 전송이 밀리고 있다는 뜻이고, 그만큼이 유실 위험 구간입니다.

샤드 간 데이터 분포가 균일한지도 확인해야 합니다.

-- 샤드별 행 수 - 편차가 크면 샤딩 키 분포 문제
SELECT hostName() AS host, count() AS rows
FROM clusterAllReplicas('my_cluster', currentDatabase(), events_local)
GROUP BY host
ORDER BY host;

clusterAllReplicas는 모든 레플리카를 도므로 복제본까지 함께 보입니다. 샤드별 값만 필요하면 cluster 함수를 씁니다. 편차가 몇 배 이상 벌어져 있으면 샤딩 키 선택(3절)을 다시 봐야 하고, 재적재가 필요하다는 뜻이므로 빨리 발견할수록 좋습니다.

8. 트러블슈팅

증상	원인	조치
노드 장애 시 일부 데이터가 조회에서 사라짐	로컬 테이블이 MergeTree라 복제 없음	ReplicatedMergeTree로 전환
레플리카 간 데이터가 다름	매크로 경로가 어긋남	system.macros, Keeper 경로 확인
같은 행이 두 번 보임	internal_replication: false + 복제 테이블	true로 변경
INSERT 성공했는데 데이터가 없음	Distributed 백그라운드 전송 중 유실	로컬 직접 INSERT 또는 동기 전송
INSERT가 막힘 (읽기는 됨)	Keeper 연결 끊김	is_readonly 확인, Keeper 복구
샤드 추가 후에도 성능 그대로	기존 데이터가 재분배되지 않음	수동 재적재 계획
특정 샤드만 디스크가 참	샤딩 키 분포 편중	해시 적용, 가중치 조정
조인이 매우 느림	조인 키와 샤딩 키가 다름	같은 키로 샤딩 또는 사전 집계
Keeper 부하가 높음	파티션이 잘게 쪼개져 파트 폭증	파티션 키 단위 확대

9. 마무리

• 샤딩과 복제는 서로 독립된 메커니즘입니다. Distributed만 만들면 나누기만 되고, ReplicatedMergeTree만 쓰면 사본만 생깁니다. 둘 다 설정해야 클러스터가 됩니다.
• Distributed는 데이터를 저장하지 않습니다. 로컬 테이블이 각 노드에 있어야 하고, 매크로 경로가 어긋나면 오류 없이 데이터가 뒤섞이거나 복제가 안 됩니다.
• 샤딩 키는 사실상 되돌릴 수 없습니다. 샤드를 늘려도 기존 데이터는 재분배되지 않습니다. 조인 지역성을 우선 기준으로 삼는 편이 안전합니다.
• ReplicatedMergeTree를 쓴다면 internal_replication은 반드시 true입니다. 아니면 두 복제 경로가 겹칩니다.
• Distributed에 INSERT하면 성공 응답 뒤에도 전송이 남아 있습니다. 대량 적재는 로컬 테이블 직접 삽입이나 재전송 가능한 원본을 두는 쪽이 맞습니다.
• Keeper 가용성이 곧 쓰기 가용성입니다. 파티션을 잘게 쪼개면 Keeper가 먼저 무너집니다.

참고

• ClickHouse - Distributed Table Engine (opens new window)
• ClickHouse - Data Replication (ReplicatedMergeTree) (opens new window)
• ClickHouse - ClickHouse Keeper (opens new window)
• ClickHouse - Distributed DDL (ON CLUSTER) (opens new window)
• ClickHouse - system.replicas (opens new window)