Kafka와 스트리밍 수집 (Kafka 엔진, ClickPipes, Kafka Connect 싱크)

세 가지 길

이 밖에 OpenTelemetry Collector 의 ClickHouse exporter, Vector, Fluent Bit, Redpanda Connect 같은 수집기도 Kafka 를 읽어 ClickHouse 로 씁니다. 어떤 길이든 끝에서는 INSERT 모범 사례 의 원칙(큰 배치, 낮은 빈도) 이 그대로 적용됩니다.

Kafka 테이블 엔진: 구조

Kafka 엔진 테이블은 데이터를 저장하지 않습니다. 토픽을 바라보는 스트림 이고, 그 위에 걸린 머티리얼라이즈드 뷰가 메시지를 꺼내 실제 MergeTree 테이블에 넣습니다. 세 조각이 한 세트입니다.

-- 1. 토픽을 바라보는 스트림 테이블
CREATE TABLE events_queue (
  ts DateTime64(3), event LowCardinality(String), user_id UInt64, props String
) ENGINE = Kafka
SETTINGS kafka_broker_list = 'kafka-1:9092,kafka-2:9092',
         kafka_topic_list = 'events',
         kafka_group_name = 'clickhouse-events',
         kafka_format = 'JSONEachRow',
         kafka_num_consumers = 2,
         kafka_max_block_size = 65536,
         kafka_flush_interval_ms = 7500,
         kafka_handle_error_mode = 'stream';

-- 2. 실제 저장 테이블
CREATE TABLE events (
  ts DateTime64(3), event LowCardinality(String), user_id UInt64, props String,
  _topic LowCardinality(String), _partition UInt32, _offset UInt64
) ENGINE = MergeTree ORDER BY (event, ts);

-- 3. 스트림에서 꺼내 저장 테이블로 넣는 뷰
CREATE MATERIALIZED VIEW events_mv TO events AS
SELECT ts, event, user_id, props, _topic, _partition, _offset
FROM events_queue WHERE _error = '';

뷰를 만든 순간부터 소비가 시작됩니다. 뷰가 없으면 메시지는 어디로도 안 갑니다. 반대로 소비를 잠깐 멈추려면 뷰를 DETACH 하거나 Kafka 테이블을 DETACH 합니다.

가상 컬럼으로 _topic, _partition, _offset, _timestamp, _key, _headers 를 쓸 수 있습니다. 오프셋을 저장 테이블에 함께 남겨 두면 나중에 "어디까지 들어왔나" 를 ClickHouse 쪽에서 바로 확인할 수 있어 운영이 편합니다.

배치와 flush

컨슈머는 kafka_max_block_size 행이 모이거나 kafka_flush_interval_ms(기본은 서버의 stream_flush_interval_ms, 7500 ms) 가 지나면 블록을 뷰에 넘깁니다. 이 블록 하나가 저장 테이블의 파트 하나입니다. 메시지 유입이 적은 토픽에서 flush 간격을 짧게 잡으면 작은 파트가 끊임없이 생기므로, 지연 요구가 느슨하다면 간격을 늘리는 쪽이 서버에 좋습니다. kafka_poll_max_batch_size, kafka_poll_timeout_ms 는 librdkafka 폴링 단위라 보통 건드리지 않습니다.

깨진 메시지 처리

kafka_handle_error_mode = 'stream' 을 켜면 파싱에 실패한 메시지도 버려지지 않고 _raw_message, _error 가상 컬럼에 담겨 뷰로 넘어옵니다. 위 예시처럼 _error = '' 인 것만 저장 테이블로 보내고, 실패한 것은 별도 뷰로 데드레터 테이블에 남기는 패턴이 표준입니다. 기본값 default 모드에서는 깨진 메시지 하나가 블록 전체를 실패시키고 소비가 멈춥니다.

CREATE TABLE events_dead (ts DateTime DEFAULT now(), raw String, error String, _offset UInt64)
ENGINE = MergeTree ORDER BY ts;

CREATE MATERIALIZED VIEW events_dead_mv TO events_dead AS
SELECT now() AS ts, _raw_message AS raw, _error AS error, _offset
FROM events_queue WHERE _error != '';

주의 뷰 쿼리 자체가 실패하면(예: 대상 테이블 컬럼 타입 불일치, 메모리 초과) 그 블록은 커밋되지 않고 같은 메시지를 다시 읽습니다. 로그에 같은 오류가 반복되며 소비가 제자리인 상태가 됩니다. system.kafka_consumers 의 exceptions 와 last_poll_time 을 보면 바로 보입니다.

보안 설정과 named collection

SASL, TLS 같은 librdkafka 옵션은 서버 설정의 <kafka> 블록에 둡니다. 토픽별로 다르게 주려면 <kafka_토픽이름> 블록을 씁니다. 브로커 주소와 자격 증명을 쿼리에 박지 않으려면 named collection 을 만들어 ENGINE = Kafka(kafka_prod) 처럼 참조합니다.

<clickhouse>
  <kafka>
    <security_protocol>SASL_SSL</security_protocol>
    <sasl_mechanism>SCRAM-SHA-512</sasl_mechanism>
    <sasl_username from_env="KAFKA_USER"/>
    <sasl_password from_env="KAFKA_PASSWORD"/>
  </kafka>
</clickhouse>

중복과 순서

컨슈머 그룹 오프셋은 블록이 뷰까지 성공한 뒤 커밋됩니다. 커밋 직전에 서버가 죽으면 같은 메시지를 다시 읽으므로 중복이 생길 수 있고, 유실은 없습니다. 중복이 문제면 _partition, _offset 을 키로 쓰는 ReplacingMergeTree 로 받거나 저장 테이블 쪽에서 블록 중복 제거 토큰을 활용합니다. MergeTree 가족 참고. 파티션 안 순서는 유지되지만 파티션 사이 순서는 보장되지 않습니다. 시간 순서가 필요한 집계는 ts 로 정렬해서 처리하세요.

Keeper 에 오프셋을 보관하는 새 Kafka 엔진 구현(kafka_keeper_path, kafka_replica_name 설정) 은 Exp 단계로, 블록 커밋과 오프셋 커밋을 한 트랜잭션처럼 묶어 중복을 줄입니다. 운영에 쓰기 전에 버전별 상태를 확인하세요.

ClickPipes (ClickHouse Cloud)

ClickHouse Cloud 에서는 Kafka 엔진 대신 ClickPipes 가 기본 선택입니다. 콘솔에서 브로커, 인증, 토픽, 포맷을 고르면 샘플을 읽어 스키마를 보여 주고 대상 테이블을 만들어 줍니다.

• 포맷: JSON(각 줄 또는 객체), Avro(Confluent 스키마 레지스트리 연동), 2026년 기준 Protobuf 도 지원합니다. 스키마 레지스트리의 변경을 따라가며 컬럼을 늘릴 수 있습니다.
• 소스: Apache Kafka, Confluent Cloud, Amazon MSK, Redpanda, WarpStream, Azure Event Hubs. 인증은 SASL/PLAIN, SCRAM, IAM(MSK), mTLS.
• 확장: 파이프마다 복제본 수와 크기를 조절하고, 유입량에 따라 스케일링합니다. 소비 지연(lag) 과 처리량, 오류를 콘솔과 system.clickpipes 관련 테이블에서 봅니다.
• 전달: at-least-once 입니다. 실패한 메시지는 데드레터 큐로 보내거나 건너뛰도록 선택합니다.
• 과금: ClickPipes 컴퓨트 시간과 수집한 데이터 양으로 따로 과금됩니다. 비용 구조와 절감 에 정리했습니다.

ClickPipes 가 만든 대상 테이블은 평범한 MergeTree 테이블이라 그 위에 머티리얼라이즈드 뷰를 걸어 집계 테이블을 만드는 방식은 똑같습니다.

Kafka Connect 싱크 (clickhouse-kafka-connect)

Confluent Hub 에도 올라와 있는 공식 커넥터입니다. Connect 워커가 토픽을 읽어 ClickHouse 에 HTTP 로 INSERT 하며, 처리한 오프셋 범위를 ClickHouse 안의 KeeperMap 엔진 테이블(connect_state) 에 기록해 정확히 한 번(exactly-once) 을 구현합니다. 같은 범위가 다시 오면 상태 테이블과 비교해 건너뜁니다.

{
  "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
  "topics": "events",
  "hostname": "xxxx.clickhouse.cloud", "port": "8443", "ssl": "true",
  "username": "default", "password": "...", "database": "default",
  "exactlyOnce": "true",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter.schemas.enable": "false",
  "errors.tolerance": "all", "errors.deadletterqueue.topic.name": "events-dlq"
}

토픽 이름과 테이블 이름이 같아야 하고(topic2TableMap 으로 매핑 가능), 스키마가 있는 Avro 나 스키마 레지스트리 JSON 을 쓰면 타입 매핑이 깔끔합니다. 배치 크기는 Connect 쪽 consumer.max.poll.records 와 fetch 설정이 정합니다.

어느 길을 고를까

1. ClickHouse Cloud 를 쓴다면 ClickPipes 로 시작합니다. 운영 부담이 가장 적습니다.
2. 셀프 호스팅이고 Connect 클러스터가 없다면 Kafka 엔진. 단, 뷰 오류로 소비가 멈추는 패턴에 대비해 데드레터와 모니터링을 처음부터 같이 만듭니다.
3. 정확히 한 번이 요건이고 Connect 가 이미 있다면 Kafka Connect 싱크.
4. 로그와 트레이스처럼 OpenTelemetry 로 모이는 데이터는 Kafka 를 거치지 말고 Collector 의 ClickHouse exporter 로 직접 넣는 쪽이 단순합니다. ClickStack 과 chDB 참고.

운영 체크리스트

-- 컨슈머 상태: 마지막 폴링, 커밋, 예외
SELECT database, table, consumer_id, assignments.topic, assignments.partition_id,
       assignments.current_offset, last_poll_time, num_messages_read, last_commit_time,
       num_commits, exceptions.text
FROM system.kafka_consumers ARRAY JOIN assignments, exceptions
ORDER BY last_poll_time DESC;

-- 저장 테이블에 남긴 오프셋으로 파티션별 진행 확인
SELECT _partition, max(_offset) AS last_offset, max(ts) AS last_ts
FROM events GROUP BY _partition ORDER BY _partition;

• lag 은 Kafka 쪽 도구(kafka-consumer-groups.sh, Burrow, Cloud 콘솔) 로 봅니다. ClickHouse 는 자기가 읽은 오프셋만 알고 토픽 끝은 모릅니다.
• 재처리 가 필요하면 뷰를 DETACH 한 뒤 컨슈머 그룹 오프셋을 Kafka 쪽에서 리셋하고 다시 ATTACH 합니다. 그룹 이름을 바꾸면 처음부터(또는 auto.offset.reset 설정대로) 읽습니다.
• 컨슈머 수 는 kafka_num_consumers 와 서버 코어 수, 토픽 파티션 수 안에서 잡습니다. 파티션보다 컨슈머가 많으면 노는 컨슈머가 생기고, 복제본 여러 대가 같은 그룹으로 붙으면 파티션이 나뉘어 자연스럽게 분산됩니다.
• 스키마 변경 은 새 컬럼이 늘어나는 방향이면 Kafka 테이블, 뷰, 저장 테이블 순서로 ALTER 하고, 그 사이 들어오는 메시지는 input_format_skip_unknown_fields 가 받아 줍니다.

공식 문서

• Kafka table engine
• Integrating Kafka with ClickHouse (개요와 선택 가이드)
• ClickPipes for Kafka
• ClickHouse Kafka Connect Sink
• system.kafka_consumers