데이터 분석 환경 구축 - 07. Clickhouse 클러스터 환경에서 데이터 삽입 및 업데이트 처리 방식 🚀(part. 1)

Clickhouse 클러스터 환경에서 데이터 삽입 및 업데이트 처리 방식 🚀(part. 1)

이전 글 데이터 분석 환경 구축 - 06. Clickhouse 테이블 종류와 용도에 이어 Kubernetes 기반 ClickHouse 클러스터 환경에서 주요 테이블 엔진별 데이터 삽입 및 업데이트 처리 방식을 살펴보고, 최신 기능과 최적화 기법을 소개합니다.

 

1. ClickHouse Kubernetes 아키텍처 개요 🛠️

Kubernetes 환경에서 ClickHouse는 샤드(Shard)와 레플리카(Replica) 구조를 통해 데이터를 분산 저장하고 고가용성을 보장합니다.

1.1 Kubernetes 클러스터 구성

• 샤드(Shard): 데이터를 분할하여 저장하는 노드
• 레플리카(Replica): 동일 데이터를 복제하여 샤드의 장애를 대비하는 백업 노드
• ZooKeeper: 메타데이터 관리 및 레플리카 동기화

예시 아키텍처

ClickHouse Cluster (2 Shards, 2 Replicas per Shard)
┌──────────────┐   ┌──────────────┐
│ Shard 1      │   │ Shard 2      │
│ Replica 1, 2 │   │ Replica 1, 2 │
└──────────────┘   └──────────────┘

 

반응형

2. 데이터 삽입(Insert) 처리 방식 🔄

2.1 Distributed + MergeTree 조합

ClickHouse 클러스터에서 데이터를 삽입할 때, 일반적으로 Distributed 엔진과 MergeTree 계열 엔진을 조합합니다.

✅ 데이터 삽입 흐름

1. 클라이언트 요청: 사용자가 Distributed 테이블에 데이터를 삽입합니다.
2. 샤딩 라우팅: Distributed 엔진이 지정된 샤딩 키를 기준으로 데이터를 각 샤드로 분배합니다.
3. 데이터 저장: 각 샤드의 MergeTree 테이블이 데이터를 저장하고 병합(merge) 작업을 수행합니다.

✅ 테이블 생성 예제

-- 샤드별 로컬 테이블
CREATE TABLE shard_table ON CLUSTER my_cluster
(
    id UInt32,
    event_time DateTime,
    event_type String
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (id, event_time);

-- 분산 테이블
CREATE TABLE distributed_table ON CLUSTER my_cluster
(
    id UInt32,
    event_time DateTime,
    event_type String
)
ENGINE = Distributed(my_cluster, default, shard_table, cityHash64(id));

✅ 데이터 삽입 예제

INSERT INTO distributed_table VALUES 
(1, '2025-04-01 10:00:00', 'login'),
(2, '2025-04-01 10:05:00', 'purchase'),
(3, '2025-04-01 10:10:00', 'logout');

 

 

⭕️ 실전 예시 - 이벤트 처리

🎯 실전 예제에서는 shard 2개를 생성하여 데이터를 처리합니다.

NAME                     READY   STATUS    RESTARTS
clickhouse-shard0-0      1/1     Running   0       
clickhouse-shard1-0      1/1     Running   0       
clickhouse-zookeeper-0   1/1     Running   0

 

1️⃣ MergeTree 엔진을 사용한 테이블에만 데이터를 넣었을 때

-- 일반 데이터베이스 생성(Replicated X)
CREATE DATABASE IF NOT EXISTS regular_db;

-- 테이블 생성
CREATE TABLE IF NOT EXISTS regular_db.test_merge_tree (
    id UInt64,
    event_date Date,
    event_time DateTime,
    user_id String,
    event_type String,
    payload String
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, id);

-- 임의 데이터 Insert
INSERT INTO regular_db.test_merge_tree
SELECT
    number as id,
    toDate('2024-03-01') + (number % 30) as event_date,
    toDateTime(toDate('2024-03-01') + (number % 30) + (number % 24) / 24) as event_time,
    concat('user_', toString(1 + number % 1000)) as user_id,
    multiIf(
        number % 5 = 0, 'click',
        number % 5 = 1, 'view',
        number % 5 = 2, 'purchase',
        number % 5 = 3, 'login',
        'logout'
    ) as event_type,
    concat('{"data":"', repeat('X', 10 + number % 50), '","value":', toString(number % 100), '}') as payload
FROM numbers(1000000);

 

샤드별 데이터 저장 위치별 크기(byte)와 row수 확인

SELECT hostName(), sum(rows) AS total_rows, sum(bytes) AS total_bytes
FROM cluster('default', 'system', 'parts')
WHERE database = 'regular_db' AND table = 'test_merge_tree'
GROUP BY hostName();

• 별도 지정 없이 mergetree에 직접 데이터를 추가했을 때 shard0-0에 추가되는 것을 알 수 있습니다.

2️⃣ MergeTree 엔진을 사용한 테이블에 Distributed 엔진을 사용한 테이블을 이용하여 데이터를 넣었을 때

• 데이터베이스, 테이블을 재 생성하고 데이터 추가 테스트를 진행합니다.

-- 데이터베이스 제거 및 재 생성
DROP DATABASE regular_db;
CREATE DATABASE IF NOT EXISTS regular_db on cluster default;

-- 테이블 재 생성
CREATE TABLE IF NOT EXISTS regular_db.test_merge_tree on cluster default (
    id UInt64,
    event_date Date,
    event_time DateTime,
    user_id String,
    event_type String,
    payload String
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, id);

-- 분산 테이블 생성
CREATE TABLE regular_db.distributed_table ON CLUSTER default
(
    id UInt64,
    event_date Date,
    event_time DateTime,
    user_id String,
    event_type String,
    payload String
)
ENGINE = Distributed(default, regular_db, test_merge_tree, cityHash64(id));

-- 분산 테이블에 데이터 추가
INSERT INTO regular_db.distributed_table
SELECT
    number as id,
    toDate('2024-03-01') + (number % 30) as event_date,
    toDateTime(toDate('2024-03-01') + (number % 30) + (number % 24) / 24) as event_time,
    concat('user_', toString(1 + number % 1000)) as user_id,
    multiIf(
        number % 5 = 0, 'click',
        number % 5 = 1, 'view',
        number % 5 = 2, 'purchase',
        number % 5 = 3, 'login',
        'logout'
    ) as event_type,
    concat('{"data":"', repeat('X', 10 + number % 50), '","value":', toString(number % 100), '}') as payload
FROM numbers(1000000);

 

샤드별 row수와 byte를 확인하면 다음과 같이 분산되어 저장된 것을 확인할 수 있습니다.

3. 결론 💡

Kubernetes 환경에서 ClickHouse를 운영하면 확장성과 고가용성을 얻을 수 있습니다. 이를 위해 적절한 테이블 엔진, 데이터베이스 엔진 을 사용하는 것이 중요합니다.

 

✅ 다음 시간에는 ReplicatedMergetree 엔진을 설명하고 Distributed + Mergetree 구성과 비교하여 설명하겠습니다. 또한 어떤 상황에서 각 조합이 사용되는지 설명합니다.