` | MAP | Map(K, V) | ✅ | |
| oneof | STRUCT | Tuple / Variant | ⚠️ | `oneof`를 ClickHouse 스키마로 변환하는 방법은 아래 섹션을 참조하십시오 |
| google.protobuf.DoubleValue | FLOAT64 | Nullable(Float64) | ✅ | |
| google.protobuf.FloatValue | FLOAT32 | Nullable(Float32) | ✅ | |
| google.protobuf.Int64Value | INT64 | Nullable(Int64) | ✅ | |
| google.protobuf.UInt64Value | INT64 | Nullable(UInt64) | ✅ | |
| google.protobuf.UInt32Value | INT64 | Nullable(UInt32) | ✅ | |
| google.protobuf.Int32Value | INT32 | Nullable(Int32) | ✅ | |
| google.protobuf.BoolValue | BOOLEAN | Nullable(Bool) | ✅ | |
| google.protobuf.StringValue | STRING | Nullable(String) | ✅ | |
| google.protobuf.BytesValue | BYTES | Nullable(String) | ✅ | |
| google.protobuf.Timestamp | org.apache.kafka.connect.data.Timestamp | DateTime64(3) | ✅ | |
| google.type.Date | org.apache.kafka.connect.data.Date | Date | ✅ | |
| google.type.TimeOfDay | org.apache.kafka.connect.data.Time | Int32 / Int64 | ✅ | |
| google.protobuf.Duration | STRUCT | Tuple(`seconds` Int64, `nano` Nullable(Int32)) | ✅ | |
| google.protobuf.Any | *N/A* | *N/A* | ❌ | |
| google.protobuf.Empty | *N/A* | *N/A* | ❌ | |
Kafka Connect 타입과 ClickHouse 타입의 매핑은 [지원되는 데이터 타입](#supported-data-types)을 참조하십시오.
#### `oneof` 필드를 ClickHouse 컬럼으로 변환할 때 참고 사항
커넥터는 Protobuf 유니온(`oneof`)을 ClickHouse Variant 타입으로 변환하는 기능을 지원하지 않습니다. 대신 `oneof` 필드를 ClickHouse 테이블 스키마에 각각의 널 허용 필드로 나열하십시오.
예시:
```protobuf theme={null}
syntax = "proto3";
package com.clickhouse.kafka.connect.proto.test;
message StringIntUnion {
oneof mixed {
string mixed_string = 2;
int32 mixed_int = 3;
}
}
```
다음과 같은 ClickHouse 테이블 정의로 변환됩니다:
```sql theme={null}
CREATE TABLE IF NOT EXISTS `StringIntUnion`
(
mixed_string Nullable(String),
mixed_int Nullable(Int32)
) ENGINE = ...;
```
#### 지원되지 않는 Protobuf 스키마
다음 Protobuf 스키마는 커넥터에서 지원되지 않습니다:
* 다중 메시지 유니온 (**CH 버전 26.1 이전**)
```protobuf theme={null}
syntax = "proto3";
package com.clickhouse.kafka.connect.proto.test;
message TwoRecords {
oneof payload {
TypeA type_a = 2;
TypeB type_b = 3;
}
// translates to Nullable(Tuple(label String)) in ClickHouse, which is unsupported
message TypeA {
string label = 1;
}
// translates to Nullable(Tuple(count Int32)) in ClickHouse, which is unsupported
message TypeB {
int32 count = 1;
}
}
```
CH 버전 26.1부터는 `allow_experimental_nullable_tuple_type=1`로 설정하면 이 스키마가 지원됩니다([이 문서 페이지](/docs/ko/reference/settings/session-settings#allow_experimental_nullable_tuple_type) 참조).
### JSON 스키마 지원
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
}
}
```
### String 컨버터 지원
커넥터는 [JSON](/docs/ko/reference/formats/JSON/JSONEachRow), [CSV](/docs/ko/reference/formats/CSV/CSV), [TSV](/docs/ko/reference/formats/TabSeparated/TabSeparated) 등 다양한 ClickHouse 포맷에서 String 컨버터를 지원합니다.
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"value.converter": "org.apache.kafka.connect.storage.StringConverter",
"customInsertFormat": "true",
"insertFormat": "CSV"
}
}
```
### 내부 버퍼링
내부 버퍼링을 사용하면 싱크 작업이 여러 번의 `poll()` 호출에서 레코드를 누적한 뒤, 이를 더 큰 배치로 묶어 ClickHouse에 플러시할 수 있습니다. 이렇게 하면 각 `poll()`이 파티션별로 작은 배치를 많이 생성하는 워크로드에서 처리량을 높일 수 있습니다.
주요 동작:
* `bufferCount`는 플러시하기 전에 버퍼링할 레코드 수를 제어합니다.
* `bufferFlushTime`은 버퍼링된 레코드를 플러시하기 전까지의 최대 대기 시간(밀리초)을 설정합니다.
* `bufferFlushTime`은 `bufferCount > 0`일 때만 유효합니다.
* `bufferCount=0` 및 `bufferFlushTime=0`이면 버퍼링이 비활성화된 상태로 유지됩니다(기본 동작).
* `exactlyOnce=true`일 때는 버퍼링이 지원되지 않습니다.
버퍼링이 exactly-once 모드와 호환되지 않는 이유:
버퍼링은 배치 경계를 변경하므로 ClickHouse 블록 중복 제거와 커넥터 오프셋 상태 머신이 제대로 동작하지 않게 됩니다.
이 문제를 해결하려면 커넥터 설정에서 `exactlyOnce=false`로 exactly-once 모드를 비활성화하거나, `bufferCount=0`으로 버퍼링을 비활성화하십시오.
예시:
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"exactlyOnce": "false",
"bufferCount": "5000",
"bufferFlushTime": "2000"
}
}
```
### 로깅
로깅은 Kafka Connect Platform에서 자동으로 제공됩니다.
로깅 대상과 포맷은 Kafka Connect [설정 파일](https://docs.confluent.io/platform/current/connect/logging.html#log4j-properties-file)을 통해 구성할 수 있습니다.
Confluent Platform을 사용하는 경우 CLI 명령을 실행하여 로그를 확인할 수 있습니다:
```bash theme={null}
confluent local services connect log
```
자세한 내용은 공식 [튜토리얼](https://docs.confluent.io/platform/current/connect/logging.html)을 참조하십시오.
### 모니터링
ClickHouse Kafka Connect는 [Java Management Extensions (JMX)](https://www.oracle.com/technical-resources/articles/javase/jmx.html)를 통해 런타임 메트릭을 제공합니다. JMX는 Kafka Connector에서 기본적으로 활성화되어 있습니다.
#### ClickHouse 전용 메트릭
이 커넥터는 다음 MBean 이름으로 사용자 지정 메트릭을 노출합니다:
```java theme={null}
com.clickhouse:type=ClickHouseKafkaConnector,name=SinkTask{id}
```
| 메트릭 이름 | 유형 | 설명 |
| ---------------------- | ---- | --------------------------------------------------- |
| `receivedRecords` | long | 수신된 레코드의 총 개수입니다. |
| `recordProcessingTime` | long | 레코드를 그룹화하고 통합된 구조로 변환하는 데 소요된 총 시간이며, 나노초 단위입니다. |
| `taskProcessingTime` | long | 데이터를 처리하고 ClickHouse에 삽입하는 데 소요된 총 시간이며, 나노초 단위입니다. |
#### Kafka 프로듀서/컨슈머 메트릭
커넥터는 데이터 흐름, 처리량, 성능을 파악할 수 있도록 표준 Kafka 프로듀서 및 컨슈머 메트릭을 제공합니다.
**토픽 수준 메트릭:**
* `records-sent-total`: 토픽으로 전송된 총 레코드 수
* `bytes-sent-total`: 토픽으로 전송된 총 바이트 수
* `record-send-rate`: 초당 전송된 레코드의 평균 수
* `byte-rate`: 초당 전송된 평균 바이트 수
* `compression-rate`: 달성된 압축률
**파티션 수준 메트릭:**
* `records-sent-total`: 파티션으로 전송된 총 레코드 수
* `bytes-sent-total`: 파티션으로 전송된 총 바이트 수
* `records-lag`: 파티션의 현재 지연
* `records-lead`: 파티션의 현재 선행 정도
* `replica-fetch-lag`: 레플리카의 지연 정보
**노드 수준 연결 메트릭:**
* `connection-creation-total`: Kafka 노드에 생성된 총 연결 수
* `connection-close-total`: 종료된 총 연결 수
* `request-total`: 노드로 전송된 총 요청 수
* `response-total`: 노드에서 수신한 총 응답 수
* `request-rate`: 초당 평균 요청 수
* `response-rate`: 초당 평균 응답 수
이러한 메트릭은 다음 항목을 모니터링하는 데 도움이 됩니다.
* **처리량**: 데이터 수집 속도 추적
* **지연**: 병목과 처리 지연 식별
* **압축**: 데이터 압축 효율 측정
* **연결 상태**: 네트워크 연결 및 안정성 모니터링
#### Kafka Connect Framework 메트릭
이 커넥터는 Kafka Connect Framework와 통합되며, 작업 수명 주기와 오류 추적을 위한 메트릭을 제공합니다.
**작업 상태 메트릭:**
* `task-count`: 커넥터의 전체 작업 수
* `running-task-count`: 현재 실행 중인 작업 수
* `paused-task-count`: 현재 일시 중지된 작업 수
* `failed-task-count`: 실패한 작업 수
* `destroyed-task-count`: 제거된 작업 수
* `unassigned-task-count`: 할당되지 않은 작업 수
작업 상태 값은 다음과 같습니다: `running`, `paused`, `failed`, `destroyed`, `unassigned`
**오류 메트릭:**
* `deadletterqueue-produce-failures`: 실패한 DLQ 쓰기 수
* `deadletterqueue-produce-requests`: 전체 DLQ 쓰기 시도 수
* `last-error-timestamp`: 마지막 오류의 타임스탬프
* `records-skip-total`: 오류로 인해 건너뛴 전체 레코드 수
* `records-retry-total`: 재시도한 전체 레코드 수
* `errors-total`: 발생한 전체 오류 수
**성능 메트릭:**
* `offset-commit-failures`: 실패한 offset commit 수
* `offset-commit-avg-time-ms`: offset commit의 평균 소요 시간
* `offset-commit-max-time-ms`: offset commit의 최대 소요 시간
* `put-batch-avg-time-ms`: Batch 처리 평균 시간
* `put-batch-max-time-ms`: Batch 처리 최대 시간
* `source-record-poll-total`: poll한 전체 레코드 수
#### 모니터링 모범 사례
1. **Consumer lag 모니터링**: 처리 병목을 식별할 수 있도록 파티션별 `records-lag`를 추적합니다
2. **오류율 추적**: 데이터 품질 문제를 감지할 수 있도록 `errors-total` 및 `records-skip-total`을 확인합니다
3. **작업 상태 관찰**: 작업이 정상적으로 실행되는지 확인할 수 있도록 작업 상태 메트릭을 모니터링합니다
4. **처리량 측정**: 수집 성능을 추적하기 위해 `records-send-rate` 및 `byte-rate`를 사용합니다
5. **연결 상태 모니터링**: 네트워크 문제를 확인하기 위해 노드 수준의 연결 메트릭을 점검합니다
6. **압축 효율 추적**: 데이터 전송을 최적화하기 위해 `compression-rate`를 사용합니다
자세한 JMX 메트릭 정의와 Prometheus 통합 방법은 [jmx-export-connector.yml](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/main/jmx-export-connector.yml) 설정 파일을 참조하십시오.
### 제한 사항
* 삭제는 지원되지 않습니다.
* 배치 크기는 Kafka Consumer 속성을 따릅니다.
* exactly-once에 KeeperMap을 사용하는 경우 오프셋이 변경되거나 되돌려지면 해당 토픽의 KeeperMap 내용을 삭제해야 합니다. (자세한 내용은 아래 문제 해결 가이드를 참조하십시오)
### 성능 튜닝 및 처리량 최적화
이 섹션에서는 ClickHouse Kafka Connect Sink의 성능 튜닝 전략을 설명합니다. 성능 튜닝은 처리량이 높은 환경을 다루거나 리소스 활용을 최적화하고 지연을 최소화해야 할 때 필수적입니다.
#### 성능 튜닝은 언제 필요합니까?
일반적으로 다음과 같은 상황에서는 성능 튜닝이 필요합니다.
* **고처리량 워크로드**: Kafka 토픽에서 초당 수백만 건의 이벤트를 처리하는 경우
* **Consumer lag**: 커넥터가 데이터 생성 속도를 따라가지 못해 lag가 계속 증가하는 경우
* **리소스 제약**: CPU, 메모리 또는 네트워크 사용량을 최적화해야 하는 경우
* **여러 토픽**: 대용량 토픽 여러 개를 동시에 소비하는 경우
* **작은 메시지 크기**: 서버 측 배칭의 이점을 얻을 수 있도록 작은 메시지를 대량으로 처리하는 경우
다음과 같은 경우에는 일반적으로 성능 튜닝이 **필요하지 않습니다**.
* 낮거나 중간 수준의 처리량(\< 10,000 messages/second)을 처리하는 경우
* 사용 사례에서 Consumer lag가 안정적이고 허용 가능한 수준인 경우
* 기본 커넥터 설정만으로도 필요한 처리량 요구 사항을 이미 충족하는 경우
* ClickHouse 클러스터가 유입되는 부하를 무리 없이 처리할 수 있는 경우
#### 데이터 흐름 이해하기
튜닝에 앞서 데이터가 커넥터를 통해 어떻게 흐르는지 이해하는 것이 중요합니다.
1. **Kafka Connect Framework**가 백그라운드에서 Kafka 토픽의 메시지를 가져옵니다
2. **커넥터는 폴링을 통해** 프레임워크의 내부 버퍼에서 메시지를 가져옵니다
3. **커넥터는 메시지를 배치로 묶어** 폴링 크기에 따라 처리합니다
4. **ClickHouse는** HTTP/S를 통해 배치된 삽입을 수신합니다
5. **ClickHouse는** 삽입을 처리합니다(동기식 또는 비동기식)
이 각 단계에서 성능을 최적화할 수 있습니다.
#### Kafka Connect 배치 크기 튜닝
첫 번째 최적화 단계는 커넥터가 Kafka에서 배치당 수신하는 데이터 양을 제어하는 것입니다.
##### fetch 설정
Kafka Connect(프레임워크)는 커넥터와 무관하게 백그라운드에서 Kafka 토픽의 메시지를 fetch합니다.
* **`fetch.min.bytes`**: 프레임워크가 값을 커넥터에 전달하기 전에 필요한 최소 데이터 양(기본값: 1 byte)
* **`fetch.max.bytes`**: 단일 요청으로 fetch할 수 있는 최대 데이터 양(기본값: 52428800 / 50 MB)
* **`fetch.max.wait.ms`**: `fetch.min.bytes` 조건이 충족되지 않으면 데이터를 반환하기 전까지 대기하는 최대 시간(기본값: 500 ms)
Confluent Cloud에서는 이러한 설정을 조정하려면 Confluent Cloud 지원 케이스를 열어야 합니다.
##### 폴링 설정
커넥터는 프레임워크의 버퍼에서 메시지를 폴링합니다.
* **`max.poll.records`**: 한 번의 폴링으로 반환되는 최대 레코드 수(기본값: 500)
* **`max.partition.fetch.bytes`**: 파티션별 최대 데이터 크기(기본값: 1048576 / 1 MB)
Confluent Cloud에서는 이러한 설정을 조정하려면 Confluent Cloud에서 지원 케이스를 접수해야 합니다.
##### 고처리량을 위한 권장 설정
ClickHouse에서 최적의 성능을 얻으려면 더 큰 배치를 사용하십시오:
```properties theme={null}
# 폴링당 레코드 수 증가
consumer.override.max.poll.records=5000
# 파티션 fetch 크기 증가 (5 MB)
consumer.override.max.partition.fetch.bytes=5242880
# 선택 사항: 더 많은 데이터를 기다리기 위한 최소 fetch 크기 증가 (1 MB)
consumer.override.fetch.min.bytes=1048576
# 선택 사항: 지연 시간이 중요한 경우 대기 시간 단축
consumer.override.fetch.max.wait.ms=300
```
위 속성을 사용하려면 워커 구성에서 `connector.client.config.override.policy=All`을 통해 클라이언트 재정의를 활성화해야 합니다. 자세한 내용은 [Kafka Connect 문서](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration)를 참조하십시오.
**중요**: Kafka Connect의 fetch 설정은 압축된 데이터를 기준으로 하며, ClickHouse는 비압축 데이터를 수신합니다. 압축률을 고려해 이러한 설정의 균형을 맞추십시오.
**트레이드오프**:
* **더 큰 배치** = 더 나은 ClickHouse 수집 성능, 더 적은 파트, 더 낮은 오버헤드
* **더 큰 배치** = 더 높은 메모리 사용량, 종단 간 지연 시간 증가 가능성
* **너무 큰 배치** = timeout, OutOfMemory 오류 또는 `max.poll.interval.ms` 초과 위험
자세한 내용: [Confluent 문서](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration) | [Kafka 문서](https://kafka.apache.org/documentation/#consumerconfigs)
#### 비동기 삽입
비동기 삽입은 커넥터가 비교적 작은 배치를 전송할 때, 또는 배칭 책임을 ClickHouse로 넘겨 수집을 한층 더 최적화하려는 경우에 매우 유용한 기능입니다.
##### async 삽입을 사용해야 하는 경우
다음과 같은 경우 async 삽입 활성화를 고려하십시오.
* **작은 배치가 많은 경우**: 커넥터가 작은 배치(\< 배치당 1000행)를 자주 전송하는 경우
* **높은 동시성**: 여러 커넥터 작업이 동일한 테이블에 쓰기 작업을 수행하는 경우
* **분산 배포**: 서로 다른 호스트에서 많은 커넥터 인스턴스를 실행하는 경우
* **파트 생성 오버헤드**: "too many parts" 오류가 발생하는 경우
* **혼합 워크로드**: 실시간 수집과 쿼리 워크로드를 함께 실행하는 경우
다음과 같은 경우에는 async 삽입을 **사용하지 마십시오**.
* 이미 큰 배치(배치당 10,000행 초과)를 제어된 빈도로 전송하는 경우
* 데이터가 즉시 표시되어야 하는 경우(쿼리에서 데이터를 즉시 확인해야 함)
* `wait_for_async_insert=0`을 사용하는 정확히 한 번 처리 의미 체계가 요구 사항과 충돌하는 경우
* 대신 클라이언트 측 배칭 개선의 이점을 얻을 수 있는 사용 사례인 경우
##### 비동기 삽입의 작동 방식
비동기 삽입이 활성화되면 ClickHouse는 다음과 같이 동작합니다:
1. 커넥터로부터 삽입 쿼리를 받습니다
2. 데이터를 메모리 버퍼에 기록합니다(즉시 디스크에 기록하지 않음)
3. 커넥터에 성공을 반환합니다 (`wait_for_async_insert=0`인 경우)
4. 다음 조건 중 하나가 충족되면 버퍼를 디스크로 플러시합니다:
* 버퍼 크기가 `async_insert_max_data_size`에 도달함(기본값: 100 MB)
* 첫 번째 삽입 후 `async_insert_busy_timeout_ms`밀리초가 경과함(기본값: 1000 ms)
* 누적된 쿼리 수가 최대치에 도달함(`async_insert_max_query_number`, 기본값: 100)
이 방식은 생성되는 파트 수를 크게 줄이고 전체 처리량을 향상시킵니다.
##### async 삽입 활성화
async 삽입 설정을 `clickhouseSettings` 구성 매개변수에 추가합니다:
```json theme={null}
{
"name": "clickhouse-connect",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
...
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
}
}
```
**주요 설정**:
* **`async_insert=1`**: 비동기 삽입을 활성화합니다
* **`wait_for_async_insert=1`** (권장): 커넥터가 확인 응답을 보내기 전에 데이터가 ClickHouse 스토리지에 플러시될 때까지 기다립니다. 전송 보장을 제공합니다.
* **`wait_for_async_insert=0`**: 버퍼링 직후 커넥터가 즉시 확인 응답을 보냅니다. 성능은 더 좋지만 플러시되기 전에 서버가 충돌하면 데이터가 손실될 수 있습니다.
##### async insert 동작 조정
async insert의 플러시 동작을 세부적으로 조정할 수 있습니다:
```json theme={null}
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=104857600,async_insert_busy_timeout_ms=1000"
```
일반적인 튜닝 매개변수:
* **`async_insert_max_data_size`** (기본값: 104857600 / 100 MB): 플러시 전 최대 버퍼 크기
* **`async_insert_busy_timeout_ms`** (기본값: 1000): 플러시 전 최대 시간(ms)
* **`async_insert_stale_timeout_ms`** (기본값: 0): 마지막 삽입 후 플러시되기까지의 시간(ms)
* **`async_insert_max_query_number`** (기본값: 100): 플러시 전 최대 쿼리 수
**트레이드오프**:
* **장점**: 파트 수 감소, 향상된 머지 성능, 낮은 CPU 오버헤드, 높은 동시성에서 더 나은 처리량
* **고려 사항**: 데이터를 즉시 조회할 수 없고, 엔드 투 엔드 지연 시간이 약간 증가합니다
* **위험 요소**: `wait_for_async_insert=0`인 경우 서버 장애 시 데이터가 손실될 수 있으며, 버퍼가 크면 메모리 사용량 압박이 발생할 수 있습니다
##### 정확히 한 번 처리 의미 체계의 async 삽입
비동기 삽입에서 `exactlyOnce=true`를 사용하는 경우:
```json theme={null}
{
"config": {
"exactlyOnce": "true",
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1"
}
}
```
**중요**: 오프셋 커밋이 데이터가 영구 저장된 후에만 수행되도록 하려면 exactly-once와 함께 항상 `wait_for_async_insert=1`을 사용하십시오.
async 삽입에 대한 자세한 내용은 [ClickHouse async 삽입 문서](/docs/ko/concepts/best-practices/selecting-an-insert-strategy#asynchronous-inserts)를 참조하십시오.
#### 커넥터 병렬성
처리량을 높이려면 병렬성을 늘리십시오:
##### 커넥터별 작업 수
```json theme={null}
"tasks.max": "4"
```
각 작업은 토픽 파티션의 일부를 처리합니다. 작업 수가 많을수록 병렬성은 높아지지만, 다음 사항에 유의해야 합니다:
* 효과적인 최대 작업 수 = 토픽 파티션 수
* 각 작업은 ClickHouse와의 자체 연결을 유지합니다
* 작업 수가 많을수록 오버헤드가 커지고 리소스 경합이 발생할 가능성이 높아집니다
**권장 사항**: `tasks.max`를 토픽 파티션 수와 같게 설정하여 시작한 다음, CPU 및 처리량 메트릭을 기준으로 조정하세요.
##### 배칭 시 파티션 무시
기본적으로 커넥터는 파티션별로 메시지를 배칭합니다. 처리량을 높이려면 파티션 간에 걸쳐 배칭할 수 있습니다.
```json theme={null}
"ignorePartitionsWhenBatching": "true"
```
\*\* 경고\*\*: `exactlyOnce=false`일 때만 사용하십시오. 이 설정을 사용하면 더 큰 배치를 생성해 처리량을 높일 수 있지만, 파티션별 순서 보장은 유지되지 않습니다.
#### 여러 고처리량 토픽
커넥터가 여러 토픽을 구독하도록 구성되어 있고, `topic2TableMap`을 사용해 토픽을 테이블에 매핑하며, 삽입 병목으로 인해 consumer lag이 발생하는 경우에는 토픽별로 커넥터를 하나씩 생성하는 방안을 고려하십시오.
이 문제가 발생하는 주된 이유는 현재 배치가 각 테이블에 [순차적으로](https://github.com/ClickHouse/clickhouse-kafka-connect/blob/578ac07e8be1a920aaa3b26e49183595c3edd04b/src/main/java/com/clickhouse/kafka/connect/sink/ProxySinkTask.java#L95-L100) 삽입되기 때문입니다.
**권장 사항**: 여러 고처리량 토픽의 경우, 병렬 삽입 처리량을 극대화하려면 토픽별로 커넥터 인스턴스를 하나씩 배포하십시오.
#### ClickHouse 테이블 엔진 고려 사항
사용 사례에 맞는 ClickHouse 테이블 엔진을 선택하십시오:
* **`MergeTree`**: 대부분의 사용 사례에 가장 적합하며, 쿼리와 삽입 성능의 균형이 좋습니다
* **`ReplicatedMergeTree`**: 고가용성을 위해 필요하며, 복제 오버헤드가 추가됩니다
* **적절한 `ORDER BY`를 사용하는 `*MergeTree`**: 쿼리 패턴에 맞게 최적화하십시오
**고려할 설정**:
```sql theme={null}
CREATE TABLE my_table (...)
ENGINE = MergeTree()
ORDER BY (timestamp, id)
SETTINGS
-- 병렬 파트 쓰기를 위한 최대 삽입 스레드 수 증가
max_insert_threads = 4,
-- 안정성을 위해 쿼럼(quorum) 삽입 허용 (ReplicatedMergeTree)
insert_quorum = 2
```
커넥터 수준 삽입 설정:
```json theme={null}
"clickhouseSettings": "insert_quorum=2,insert_quorum_timeout=60000"
```
#### 연결 풀링 및 타임아웃
이 커넥터는 ClickHouse와의 HTTP 연결을 유지합니다. 지연 시간이 큰 네트워크에서는 타임아웃을 조정하십시오:
```json theme={null}
"clickhouseSettings": "socket_timeout=300000,connection_timeout=30000"
```
* **`socket_timeout`** (기본값: 30000 ms): 읽기 작업의 최대 대기 시간
* **`connection_timeout`** (기본값: 10000 ms): 연결을 설정하는 최대 대기 시간
대용량 배치에서 timeout 오류가 발생하면 이 값을 늘리십시오.
#### 성능 모니터링 및 문제 해결
다음 핵심 메트릭을 모니터링하세요:
1. **Consumer lag**: Kafka 모니터링 도구를 사용해 파티션별 지연을 추적합니다
2. **커넥터 메트릭**: JMX를 통해 `receivedRecords`, `recordProcessingTime`, `taskProcessingTime`를 모니터링합니다([모니터링](#monitoring) 참조)
3. **ClickHouse 메트릭**:
* `system.asynchronous_inserts`: async insert 버퍼 사용량을 모니터링합니다
* `system.parts`: 머지 문제를 감지할 수 있도록 파트 수를 모니터링합니다
* `system.merges`: 진행 중인 머지를 모니터링합니다
* `system.events`: `InsertedRows`, `InsertedBytes`, `FailedInsertQuery`를 추적합니다
**일반적인 성능 문제**:
| 증상 | 가능한 원인 | 해결 방법 |
| ------------------- | ------------------ | ----------------------------------------------------- |
| Consumer lag 증가 | 배치가 너무 작음 | `max.poll.records`를 늘리고 async 삽입을 활성화합니다 |
| "Too many parts" 오류 | 작고 빈번한 삽입 | async 삽입을 활성화하고 배치 크기를 늘립니다 |
| timeout 오류 | 배치 크기가 큼, 네트워크가 느림 | 배치 크기를 줄이고 `socket_timeout`을 늘린 다음 네트워크를 확인합니다 |
| 높은 CPU 사용량 | 작은 파트가 너무 많음 | async 삽입을 활성화하고 머지 설정을 높입니다 |
| OutOfMemory 오류 | 배치 크기가 너무 큼 | `max.poll.records`, `max.partition.fetch.bytes`를 줄입니다 |
| 작업 부하 불균형 | 파티션 분배가 고르지 않음 | 파티션을 리밸런싱하거나 `tasks.max`를 조정합니다 |
#### 모범 사례 요약
1. **기본값으로 시작**한 다음 실제 성능을 측정하고 그 결과에 따라 조정하세요
2. **더 큰 배치를 우선하세요**: 가능하면 한 번의 삽입당 10,000\~100,000개 행을 목표로 하세요
3. **작은 배치를 많이 보내거나 동시성(Concurrency)이 높을 때는 async 삽입을 사용하세요**
4. **정확히 한 번 처리 의미 체계를 위해 `wait_for_async_insert=1`을 항상 사용하세요**
5. **수평 확장하세요**: `tasks.max`를 파티션 수만큼 늘리세요
6. **처리량을 최대화하려면 대용량 토픽마다 커넥터를 하나씩 사용하세요**
7. **지속적으로 모니터링하세요**: consumer lag, part 개수, 머지 활동을 추적하세요
8. **충분히 테스트하세요**: 프로덕션 배포 전에 반드시 실제와 유사한 부하에서 구성 변경을 테스트하세요
#### 예시: 고처리량 구성
다음은 고처리량에 맞게 최적화한 전체 예시입니다:
```json theme={null}
{
"name": "clickhouse-high-throughput",
"config": {
"connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
"tasks.max": "8",
"topics": "high_volume_topic",
"hostname": "my-clickhouse-host.cloud",
"port": "8443",
"database": "default",
"username": "default",
"password": "",
"ssl": "true",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false",
"exactlyOnce": "false",
"ignorePartitionsWhenBatching": "true",
"consumer.override.max.poll.records": "10000",
"consumer.override.max.partition.fetch.bytes": "5242880",
"consumer.override.fetch.min.bytes": "1048576",
"consumer.override.fetch.max.wait.ms": "500",
"clickhouseSettings": "async_insert=1,wait_for_async_insert=1,async_insert_max_data_size=16777216,async_insert_busy_timeout_ms=1000,socket_timeout=300000"
}
}
```
위 커넥터 구성에서는 worker 구성에서 `connector.client.config.override.policy=All`을 설정해 클라이언트 재정의를 활성화해야 합니다. 자세한 내용은 [Kafka Connect 문서](https://docs.confluent.io/platform/current/connect/references/allconfigs.html#override-the-worker-configuration)를 참조하십시오.
**이 구성은**:
* 폴링 한 번에 최대 10,000개의 레코드를 처리합니다
* 더 큰 삽입을 위해 여러 파티션의 데이터를 배치로 묶습니다
* 16 MB 버퍼를 사용하는 async 삽입을 사용합니다
* 8개의 병렬 작업을 실행합니다(파티션 수에 맞춰 설정)
* 엄격한 순서 보장보다 처리량에 중점을 두고 최적화되었습니다
### 문제 해결
#### "토픽 `[someTopic]` 파티션 `[0]`의 상태가 일치하지 않음"
이 오류는 KeeperMap에 저장된 오프셋과 Kafka에 저장된 오프셋이 서로 다를 때 발생하며, 보통 토픽이 삭제되었거나
오프셋이 수동으로 조정된 경우에 나타납니다.
이 문제를 해결하려면 해당 토픽 + 파티션에 저장된 기존 값을 삭제해야 합니다:
```sql theme={null}
-- 먼저, 데이터를 저장하는 데 사용된 데이터베이스를 확인합니다.
SELECT * FROM [database].connect_state
-- 해당 토픽과 파티션에 해당하는 키를 확인합니다.
ALTER TABLE [database].connect_state DELETE WHERE key = [keyname]
```
이 조정은 exactly-once 동작에 영향을 줄 수 있습니다.
#### "커넥터는 어떤 오류를 재시도합니까?"
현재는 일시적이어서 재시도할 수 있는 오류를 식별하는 데 중점을 두고 있으며, 여기에는 다음이 포함됩니다:
* `ClickHouseException` - ClickHouse에서 발생할 수 있는 일반적인 예외입니다.
일반적으로 서버에 과부하가 걸렸을 때 발생하며, 특히 다음 오류 코드는 일시적인 오류로 간주됩니다:
* 3 - UNEXPECTED\_END\_OF\_FILE
* 107 - FILE\_DOESNT\_EXIST
* 159 - TIMEOUT\_EXCEEDED
* 164 - READONLY
* 202 - TOO\_MANY\_SIMULTANEOUS\_QUERIES
* 203 - NO\_FREE\_CONNECTION
* 209 - SOCKET\_TIMEOUT
* 210 - NETWORK\_ERROR
* 241 - MEMORY\_LIMIT\_EXCEEDED
* 242 - TABLE\_IS\_READ\_ONLY
* 252 - TOO\_MANY\_PARTS
* 285 - TOO\_FEW\_LIVE\_REPLICAS
* 319 - UNKNOWN\_STATUS\_OF\_INSERT
* 425 - SYSTEM\_ERROR
* 999 - KEEPER\_EXCEPTION
* `SocketTimeoutException` - 소켓 timeout이 발생하면 이 예외가 발생합니다.
* `UnknownHostException` - 호스트를 확인할 수 없을 때 이 예외가 발생합니다.
* `IOException` - 네트워크에 문제가 있을 때 이 예외가 발생합니다.
#### "모든 데이터가 비어 있거나 0입니다"
데이터의 필드가 테이블의 필드와 일치하지 않을 가능성이 높습니다. 이는 특히 CDC(및 Debezium 포맷)에서 자주 발생합니다.
일반적인 해결 방법 중 하나는 connector 구성에 flatten 변환을 추가하는 것입니다:
```properties theme={null}
transforms=flatten
transforms.flatten.type=org.apache.kafka.connect.transforms.Flatten$Value
transforms.flatten.delimiter=_
```
이렇게 하면 데이터가 중첩된 JSON에서 평탄화된 JSON으로 변환됩니다(`_`를 구분자(Delimiter)로 사용). 그러면 테이블(table)의 필드는 "field1\_field2\_field3" 포맷을 따르게 됩니다(예: "before\_id", "after\_id" 등).
#### "ClickHouse에서 Kafka 키를 사용하고 싶습니다"
기본적으로 Kafka 키는 value 필드에 저장되지 않지만, `KeyToValue` 변환을 사용하면 키를 value 필드의 새 `_key` 필드로 옮길 수 있습니다:
```properties theme={null}
transforms=keyToValue
transforms.keyToValue.type=com.clickhouse.kafka.connect.transforms.KeyToValue
transforms.keyToValue.field=_key
```
