[Clickhouse] 분산 클러스터

#Clickhouse

Clickhouse는 단일 장비로도 쓸만하지만, 진짜 강점은 대규모로 분산된 빅데이터를 관리할때 빛난다.
다른 분산시스템처럼 멀티노드 대규모 환경을 전체로 만들어졌고, 초대형 데이터셋을 여러 장비에 나눠서 관리하는 것에 특화되어있다.

여기서는 샤드/레플리카의 기본적인 구조와 사용방법을 정리해본다.

기본 논리

골자는 클러스터링을 제공하는 다른 DB들과 크게 다르지 않다.
"샤드"와 "레플리카"를 통해 고가용성과 데이터 분할을 가져갈 수 있다.

https://tech.kakaopay.com/post/pallas-v2-log-platform/
큰 데이터를 쪼개서 여러 장비에 나눠 저장하는 것을 샤드라고 하고,
각 샤드를 복제해서 copy를 장비별로 분산하는 것을 Replica라고 한다.

전자는 Distributed라는 테이블 엔진을 통해 지원되며, 후자는 ReplicatedMergeTree라는 엔진을 통해 지원된다. 그리고 ReplicatedMergeTree는 keeper라는 클러스터 관리자를 통해 관리된다.

그림으로 그려보면, 분산 클러스터는 이런 식으로 구성된다.

Clickhouse는 서로를 인식하고 직접 통신한다. Clickhouse 노드끼리는 데이터를 내려받아서 데이터를 복제/분할하거나 쿼리 조합에 사용할 수 있다.
복제 과정에서 keeper란 것이 사용된다.

clickhouse keeper

그러면 저 keeper란 녀석은 무엇일까? Replication(복제)를 위한 중간 저장소라고 보면 된다.
Clickhouse 노드가 쓰기를 수행할때마다 항상 keeper에 메타데이터를 저장하고, 다른 노드들이 그걸 확인해서 복제할 것이 있다면 해당 노드를 찾아서 복제할 데이터를 다운받는 식으로 흘러가는 것이다.

유의할 점은, Shard에는 관여하지 않는다는 것이다. Shard(Distributed)는 오롯이 clickhouse의 관리 책임이다. keeper는 Replication에만 사용된다.

keeper는 단일서버로 띄워도 되긴 하지만, 보통 3 노드 이상으로 띄워서 고가용성 클러스터를 따로 구성하는 것을 권장한다. 그리고 clickhouse 노드와 별도 장비에 배치하는 것을 권한다.

https://clickhouse.com/docs/ko/architecture/cluster-deployment

그리고 keeper는 높은 성능이 필요하진 않다. 메모리는 노드당 4GB 정도만 있어도 된다.

Query 분산

그럼 사용자가 쿼리를 날릴 때는 어떻게 해야할까?
일단, 아무 clickhouse 노드에나 찔러도 동작은 한다. 쿼리를 받은 노드가 일종의 coordinator가 돼서, 부족한 것이 있다면 다른 노드까지 확인하고 결과를 조합하기 때문이다.

하지만 이 방법에는 한계가 있다. 이러면 저 쿼리를 받는 A 노드에만 부하가 가중되기 때문이다.

그래서 보통 중소규모에서는 Clickhouse 앞단에 로드밸런싱 계층을 추가해서 적절히 분산되도록 유도한다.

그리고 정말 대규모로 가면, 데이터 노드에서 읽기까지 처리하는 것이 부담스러워진다.
이 경우에는 데이터 노드와 쿼리 노드를 분리해서, Distributed 테이블만 있는 쿼리 노드를 따로 운영한다.

이런 식이다.
Query Node는 데이터를 갖고 있지 않고 Distributed 메타데이터만 들고 분산 쿼리를 처리한다.

클러스터 시작해보기 (docker compose)

로컬에서 단일 장비로 동작을 확인해보겠다.
전체 예제는 여기에 있다.
https://github.com/myyrakle/infrastructures/tree/master/docker-compose/clickhouse-cluster

적당히 compose를 실행하면 바로 시작될 것이다.

그리고 각각의 clickhouse 노드에 접근해보자.

그럼 이제 분산 구성을 활용해서 이것저것 실험해볼 수 있을 것이다.

샤드 구성해보기 (파티션)

먼저 샤드를 구성해서 하나의 데이터셋을 여러개의 노드로 분산해보자.

일단 좀 번거로운게 있는데, 노드마다 들어가서 로컬 테이블을 각자 만들어줘야 한다.

CREATE TABLE logs_local
(
    time DateTime,
    user_id UInt64,
    message String
)
ENGINE = MergeTree
ORDER BY (user_id, time);

그리고 저걸 하나로 엮여주는 Distributed 테이블을 생성한다.

CREATE TABLE logs
AS logs_local
ENGINE = Distributed(
    클러스터명,
    default,
    logs_local, -- 테이블
    cityHash64(user_id) -- 해시 키
);

그럼 이제 하나의 샤드 테이블로 구성이 된 것이다.
좀 번거롭긴 한데.

그러면 이제 적당히 Distributed 테이블을 통해 mutation을 날리면, 알아서 분산해서 저장을 해준다.

INSERT INTO logs
SELECT
    now(),
    number % 10,
    concat('log-', toString(number))
FROM numbers(1000);

그럼 각각의 로컬 테이블에 해시키 기반으로 분산된 채로 저장되었을 것이다.

내 경우에는 8:2였다.

그리고 Distributed 테이블을 통해서 조회하면 양쪽 샤드를 결합해서 보여준다.

여기서 shard+replica를 동시에 구성하려면, MergeTree를 후술할 ReplicatedMergeTree로 바꾸기만 하면 된다.

복제 (Replication)

읽기 부하의 분산이나 고가용성-데이터 방어를 원한다면, 복제까지도 구성을 해야한다.
이건 keeper가 필요하고, ReplicatedMergeTree라는 MergeTree의 변종을 사용해야 한다.

각각의 노드마다 다음과 같은 DDL을 날린다.

CREATE table logs_replicated
(
    id UInt64,
    message String
)
ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/logs_replicated', -- 복제 그룹
    'ch1' -- 노드 ID
)

CREATE TABLE logs_replicated
(
    id UInt64,
    message String
)
ENGINE = ReplicatedMergeTree(
    '/clickhouse/tables/logs_replicated', -- 복제 그룹
    'ch2' -- 노드 ID
)

그러면 이제 한 곳에만 데이터를 부어도

모든 노드에 자동으로 복제될 것이다.

분산 DDL

근데 노드마다 각각 들어가서 테이블 추가하고, 컬럼 추가하는건 너무 번거롭다.
다행히도 이런 번거로움을 덜어줄 수 있는 분산 DDL 기능을 제공한다. 테이블명 오른편에 클러스터명을 넣어주기만 하면 알아서 모든 노드에 DDL까지 복제를 해준다.

CREATE table asdf ON CLUSTER my_cluster
(
    time DateTime,
    message String
)
ENGINE = MergeTree
ORDER BY (time);

그럼 양쪽에 다 추가되어있을 것이다.

참고로 이 기능은 keeper에 의존한다.

참조
https://clickhouse.com/docs/ko/architecture/cluster-deployment
https://clickhouse.com/docs/ko/guides/sre/keeper/clickhouse-keeper
https://clickhouse.com/docs/ko/sql-reference/distributed-ddl
https://clickhouse.com/docs/engines/table-engines/special/distributed
https://tech.kakaopay.com/post/pallas-v2-log-platform/
https://clickhouse.com/docs/ko/deployment-guides/parallel-replicas