복제와 샤딩 (ReplicatedMergeTree, Keeper, Distributed, 병렬 복제본)

복제와 샤딩은 다른 문제

• 복제(replication): 같은 데이터를 여러 서버에 복사해 가용성과 읽기 처리량을 얻습니다. ReplicatedMergeTree가 담당합니다.
• 샤딩(sharding): 데이터를 서버마다 다른 부분으로 나누어 한 서버에 안 들어가는 용량과 쓰기를 분산합니다. Distributed 테이블이 담당합니다.

둘을 조합한 "샤드 N개 x 복제본 M개"가 전통적 셀프 호스팅 클러스터입니다. 다만 하드웨어가 커지고 병렬 복제본과 오브젝트 스토리지 기반 엔진이 들어오면서, 샤딩 없이 복제본만 늘리는 구성이 요즘의 기본 권장입니다. 샤딩은 복잡성을 크게 올리므로 꼭 필요할 때만 합니다.

Keeper: 복제의 조정자

복제 테이블은 데이터 자체가 아니라 "어떤 파트가 있어야 하는가"라는 로그를 Keeper(또는 ZooKeeper)에 기록합니다. 각 복제본은 로그를 따라가며 자기에게 없는 파트를 다른 복제본에서 가져옵니다(fetch). 머지도 한 복제본이 결정해 로그에 올리고, 다른 복제본은 같은 머지를 따라 하거나 결과를 가져옵니다.

• ClickHouse Keeper는 ZooKeeper 프로토콜 호환 구현으로, 지금은 ZooKeeper 대신 Keeper를 권장합니다. 3노드(또는 5노드) 홀수 구성.
• Keeper는 메타데이터만 다루므로 작은 서버로도 충분하지만, 지연에 민감합니다. 디스크는 SSD, ClickHouse 서버와 같은 노드에 두는 것은 소규모에서만.
• Keeper가 멈추면 복제 테이블은 읽기 전용이 됩니다(읽기는 됨). INSERT는 Table is in readonly mode 오류로 실패합니다. 복제 지연과 Keeper 문제에서 다룹니다.

<!-- config.xml 일부: Keeper 접속과 매크로 -->
<zookeeper>
    <node><host>keeper-1</host><port>9181</port></node>
    <node><host>keeper-2</host><port>9181</port></node>
    <node><host>keeper-3</host><port>9181</port></node>
</zookeeper>
<macros>
    <cluster>prod</cluster>
    <shard>01</shard>
    <replica>ch-1</replica>
</macros>

ReplicatedMergeTree 만들기

CREATE TABLE events ON CLUSTER prod
(
    ts DateTime, tenant_id UInt32, event LowCardinality(String), user_id UInt64
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/events', '{replica}')
ORDER BY (tenant_id, ts);

• 첫 인수는 Keeper 안의 경로입니다. 같은 샤드의 복제본은 같은 경로, 다른 샤드는 다른 경로여야 합니다. 매크로로 서버마다 알맞게 치환됩니다.
• 두 번째 인수는 복제본 이름으로 샤드 안에서 유일해야 합니다.
• database_replicated_allow_... 설정과 Replicated 데이터베이스 엔진을 쓰면 테이블마다 경로를 적지 않아도 되고 DDL이 자동으로 모든 복제본에 전파됩니다. 새 클러스터는 Replicated 데이터베이스를 권장합니다.

CREATE DATABASE analytics ENGINE = Replicated('/clickhouse/databases/analytics', '{shard}', '{replica}');
-- 이 DB 안에서는 ENGINE = ReplicatedMergeTree 만 쓰면 경로가 자동

복제가 하는 일과 하지 않는 일

• INSERT는 어느 복제본에 해도 됩니다. 쓰기 리더가 없습니다. 복제는 비동기이고, 기본값에서 INSERT는 한 복제본에 쓰이면 성공을 돌려줍니다. insert_quorum = 2로 두 복제본에 쓰여야 성공으로 보게 할 수 있습니다.
• 읽기는 복제본에 따라 잠시 결과가 다를 수 있습니다(최종 일관성). 방금 쓴 데이터를 바로 읽어야 하면 같은 복제본에 쿼리하거나 select_sequential_consistency = 1(Cloud에서 특히 의미 있음)을 씁니다.
• INSERT 블록 중복 제거: 같은 데이터 블록(체크섬 기준, 또는 insert_deduplication_token)을 다시 넣으면 무시됩니다. 재시도를 안전하게 하는 장치입니다. INSERT 모범 사례.

Distributed 테이블로 샤딩

Distributed 엔진은 데이터를 갖지 않는 프록시 테이블입니다. 읽으면 모든 샤드에 쿼리를 보내 결과를 합치고, 쓰면 샤딩 키로 행을 나누어 각 샤드로 보냅니다.

<remote_servers>
    <prod>
        <shard>
            <internal_replication>true</internal_replication>
            <replica><host>ch-1a</host><port>9000</port></replica>
            <replica><host>ch-1b</host><port>9000</port></replica>
        </shard>
        <shard>
            <internal_replication>true</internal_replication>
            <replica><host>ch-2a</host><port>9000</port></replica>
            <replica><host>ch-2b</host><port>9000</port></replica>
        </shard>
    </prod>
</remote_servers>

CREATE TABLE events_dist ON CLUSTER prod AS events
ENGINE = Distributed(prod, currentDatabase(), events, cityHash64(tenant_id));

• internal_replication = true가 핵심입니다. Distributed가 각 샤드의 복제본 하나에만 쓰고 복제는 ReplicatedMergeTree에 맡깁니다. false면 모든 복제본에 따로 써서 데이터가 어긋날 수 있습니다.
• 샤딩 키는 쿼리에서 자주 묶는 열(테넌트)로 잡으면 GROUP BY tenant_id를 샤드 안에서 끝낼 수 있습니다(distributed_group_by_no_merge, optimize_skip_unused_shards).
• Distributed로 쓰기는 각 샤드로 보내기 전에 로컬 디스크에 임시 저장(비동기)되며 system.distribution_queue에 쌓입니다. 운영에서는 애플리케이션이 샤드를 직접 골라 로컬 테이블에 쓰는 쪽이 단순하고 안전합니다.
• 조인은 샤드 간에 어렵습니다. 작은 테이블은 모든 샤드에 복제해 두고, 큰 테이블끼리는 같은 샤딩 키로 colocated 시키거나 GLOBAL JOIN으로 오른쪽 테이블을 모든 샤드에 보냅니다.

주의 샤드를 추가하면 기존 데이터는 자동으로 재배치되지 않습니다. 리밸런싱은 직접 해야 하고(파티션 단위 복사), 샤딩 키 기반 최적화도 그동안 틀어집니다. 이 비용 때문에 "지금 샤딩이 꼭 필요한가"를 먼저 묻습니다. 서버 한 대가 수십 TB를 다루는 것이 요즘은 보통입니다.

병렬 복제본: 샤딩 없이 읽기 나누기

병렬 복제본(parallel replicas)은 한 샤드의 여러 복제본이 같은 쿼리를 그래뉼 범위로 나눠 읽는 기능입니다. 데이터는 전부 복제되어 있으니 리밸런싱이 없고, 복제본을 늘리면 읽기 처리량이 그만큼 늘어납니다. 25.x에서 GA가 되었고 ClickHouse Cloud에서는 기본 동작 가운데 하나입니다.

SET enable_parallel_replicas = 1, max_parallel_replicas = 3, cluster_for_parallel_replicas = 'prod';
SELECT tenant_id, count() FROM events WHERE ts >= today() - 30 GROUP BY tenant_id;

• 코디네이터 복제본이 작업을 나누어 주고, 각 복제본이 일부를 읽어 중간 결과를 보냅니다. 큰 스캔과 집계에 효과가 크고, 작은 쿼리는 오히려 조정 비용이 듭니다. parallel_replicas_min_number_of_rows_per_replica로 작은 쿼리는 제외합니다.
• 복제본끼리 데이터가 같아야 정확합니다. 복제 지연이 큰 복제본이 있으면 그 복제본은 제외되거나 결과에 영향을 줄 수 있습니다.

Cloud의 SharedMergeTree

ClickHouse Cloud는 ReplicatedMergeTree 대신 SharedMergeTree를 씁니다. 데이터는 오브젝트 스토리지(S3, GCS, Azure Blob)에 한 번만 저장되고, 모든 컴퓨트 노드가 그것을 공유합니다. 노드끼리 파트를 가져오는 fetch가 없으므로 복제본 추가가 거의 즉시이고, 머지도 한 노드가 하면 모두에게 반영됩니다. 사용자가 ENGINE = MergeTree라고 적어도 내부적으로 SharedMergeTree로 만들어집니다.

• Keeper는 Cloud가 운영하고, 파트 메타데이터만 조정합니다.
• 노드 사이 지연이 있으므로 방금 쓴 데이터를 다른 노드에서 바로 읽으려면 select_sequential_consistency = 1을 씁니다.
• 자세한 구조와 compute-compute 분리(웨어하우스)는 ClickHouse Cloud 개요와 컴퓨트와 스케일링.

운영 쿼리

-- 복제 상태: 큐 길이, 지연, 읽기 전용 여부
SELECT database, table, is_readonly, absolute_delay, queue_size, inserts_in_queue, merges_in_queue,
       last_queue_update_exception
FROM system.replicas
WHERE absolute_delay > 60 OR is_readonly OR queue_size > 100;

-- 복제 큐 상세
SELECT type, create_time, num_tries, last_exception, is_currently_executing
FROM system.replication_queue ORDER BY create_time LIMIT 20;

-- 클러스터 정의 확인
SELECT cluster, shard_num, replica_num, host_name, port, is_local FROM system.clusters WHERE cluster = 'prod';

-- 모든 복제본에서 같은 쿼리 (clusterAllReplicas)
SELECT hostName(), count() FROM clusterAllReplicas('prod', system.parts) WHERE active GROUP BY 1;

구성 가이드 요약

공식 문서

• Data replication
• ClickHouse Keeper
• Distributed table engine
• Replicated database engine
• Parallel replicas
• SharedMergeTree