Kafka Engine ClickHouse: 손실 없는 원자성

ClickHouse의 Kafka Engine: 데이터 손실 없이 원자적 삽입 보장

MergeTree 테이블에 빈번한 개별 삽입(초당 10,000 INSERT)이 발생하면 성능이 초당 59행으로 떨어집니다. ReplicatedMergeTree의 경우 초당 6행으로, 쿼럼을 사용하면 초당 2행에 불과합니다. 이는 max_merge_selecting_sleep_ms 설정에 따라 주기적으로 병합되는 수많은 작은 데이터 파트 생성 때문입니다. 병합 프로세스가 CPU와 I/O를 과부하시켜 전체 속도를 저하시킵니다.

Buffer 테이블은 해결책처럼 보입니다: 데이터가 RAM에 축적되어 시간, 크기 또는 행 수에 따라 배치로 대상 테이블에 플러시됩니다. 이로 인해 파일 수와 병합 부하가 최소화됩니다. 그러나 ClickHouse가 크래시되면 메모리에 플러시되지 않은 데이터가 손실됩니다.

Buffer 테이블: 메커니즘과 위험

Buffer 테이블은 Buffer 엔진을 사용하며, 플러시 조건이 충족될 때까지 메모리에 데이터를 축적합니다: max_time_to_flush (예: 100초), min_time_to_flush, max_rows_to_flush (100만 행), max_bytes_to_flush (1 GB). 플러시는 MergeTree로 원자적으로 발생하지만 디스크에 쓰인 후에만입니다.

주요 위험: 서버가 크래시되면 (kill -9, OOM), RAM의 버퍼 내용이 영구적으로 손실됩니다. 버퍼에는 복제나 WAL이 없어 고부하 스트림에서 이 접근 방식이 신뢰할 수 없습니다.

Kafka Engine으로 전환하면 문제가 해결됩니다: 데이터는 Kafka에 지속적으로 저장되며, ClickHouse는 성공적인 처리 후에만 오프셋을 커밋합니다.

Kafka Engine 파이프라인 설정

간단한 스택: Docker에서 Kafka + ZooKeeper + ClickHouse. JSON 메시지로 test_topic_json 토픽을 생성합니다:

{"message": "qq"}
{"message": "bb"}

ClickHouse에서 파이프라인은 세 구성 요소로 이루어집니다:

• 읽기를 위한 Kafka Engine 테이블:

drop table if exists test_kafka_json_parse;
create table test_kafka_json_parse
(
    message String
)
engine = Kafka
SETTINGS kafka_broker_list = 'kafka:29092'
    , kafka_topic_list = 'test_topic_json'
    , kafka_group_name = 'my-group4'
    , kafka_format = 'JSONEachRow'
    , kafka_max_block_size = 2097152;

• 소비자를 위한 Materialized view:

drop view if exists test_kafka_json_parse_view;
create materialized view test_kafka_json_parse_view
TO default.message_json_parse
(
 `message` String
)
AS SELECT message FROM test_kafka_json_parse;

• 대상 MergeTree:

create table message_json_parse
(
    message String
)
engine = MergeTree
order by message;

원자성 데모

메시지 전송 후 my-group4의 소비자 오프셋은 2 (End=2)입니다. 이제 message_json_parse의 컬럼 타입을 UInt64로 변경 — JSON 문자열 "gg" 파싱이 실패합니다:

drop table if exists message_json_parse;
create table message_json_parse
(
    message UInt64
)
engine = MergeTree
order by message;

{"message": "gg"}를 전송합니다. 오프셋은 2로 유지 (End=3), ClickHouse 로그: Cannot parse string 'gg' as UInt64. 오프셋은 MV → MergeTree로 성공적인 쓰기 후에만 커밋됩니다.

ClickHouse는 메시지를 RAM에 읽지만 오류 시 오프셋을 커밋하지 않습니다. 새 메시지는 수정될 때까지 차단됩니다. 타입을 String으로 다시 수정한 후:

• "gg" 메시지가 삽입됩니다.
• 오프셋이 3으로 업데이트됩니다.
• ClickHouse 재시작 후에도 데이터 손실이 없습니다.

주요 포인트

• 원자성: Materialized view로 삽입 후 Kafka 오프셋이 커밋되어 exactly-once 의미론을 보장합니다.
• 복구: 실패/오류 후 마지막 커밋 오프셋부터 처리 재개, 중복이나 누락 없이.
• 성능: Kafka로부터 배치 처리로 MergeTree의 작은 파트 최소화.
• 신뢰성: Kafka가 데이터를 지속적으로 저장; ClickHouse는 stateless 소비자 역할.
• 제한: Kafka 인프라 필요; 엄격한 처리 위해 kafka_skip_broken_messages=0 설정.

구현 권장사항

• 병렬 처리를 위해 kafka_num_consumers >1 사용.
• 배칭을 위해 max_block_size=2MB+ 설정.
• 오프셋 모니터링: SELECT * FROM system.kafka_consumers.
• JSON의 경우: 엄격한 파싱과 함께 JSONEachRow.

— Editorial Team