%20FORMAT%20JSONEachRow`. **참고**: 쿼리는 인코딩해야 합니다.
* `Endpoint Authentication type` - BASIC
* `Auth username` - ClickHouse 사용자 이름
* `Auth password` - ClickHouse 비밀번호
이 HTTP Url은 오류가 발생하기 쉽습니다. 문제를 방지하려면 이스케이프 처리를 정확히 하십시오.
* 구성
* `Input Kafka record value format` 소스 데이터에 따라 다르지만, 대부분의 경우 JSON 또는 Avro를 사용합니다. 아래 설정에서는 `JSON`을 가정합니다.
* `advanced configurations` 섹션에서:
* `HTTP Request Method` - POST로 설정합니다
* `Request Body Format` - json
* `Batch batch size` - ClickHouse 권장 사항에 따라 **최소 1000**으로 설정합니다.
* `Batch json as array` - true
* `Retry on HTTP codes` - 400-500으로 설정하되 필요에 따라 조정하십시오. 예를 들어 ClickHouse 앞단에 HTTP 프록시가 있는 경우 달라질 수 있습니다.
* `Maximum Reties` - 기본값(10)이 적절하지만, 재시도 안정성을 높이려면 조정해도 됩니다.
HTTP Sink로 구성한 토픽에 메시지를 생성합니다
그리고 생성된 메시지가 ClickHouse 인스턴스에 기록되었는지 확인합니다.
### 문제 해결
#### HTTP Sink가 메시지를 배칭하지 않는 경우
[Sink 문서](https://docs.confluent.io/kafka-connectors/http/current/overview.html#http-sink-connector-for-cp)에 따르면 다음과 같습니다.
> Kafka header 값이 서로 다른 메시지가 포함된 경우 HTTP Sink 커넥터는 요청을 배칭하지 않습니다.
1. Kafka 레코드의 key가 동일한지 확인하십시오.
2. HTTP API URL에 매개변수를 추가하면 각 레코드마다 고유한 URL이 생성될 수 있습니다. 따라서 추가 URL 매개변수를 사용하면 배칭이 비활성화됩니다.
#### 400 잘못된 요청
##### CANNOT\_PARSE\_QUOTED\_STRING
JSON 객체를 `String` 컬럼에 삽입할 때 HTTP Sink가 다음 메시지와 함께 실패하는 경우:
```response theme={null}
Code: 26. DB::ParsingException: Cannot parse JSON string: expected opening quote: (while reading the value of key key_name): While executing JSONEachRowRowInputFormat: (at row 1). (CANNOT_PARSE_QUOTED_STRING)
```
URL에서 `input_format_json_read_objects_as_strings=1` 설정을 인코딩된 문자열 `SETTINGS%20input_format_json_read_objects_as_strings%3D1`로 지정하십시오
### GitHub 데이터셋 로드(선택 사항)
이 예시에서는 GitHub 데이터셋의 배열 필드를 그대로 유지합니다. 예시에서는 비어 있는 `github` topic이 있다고 가정하고, Kafka에 메시지를 삽입하기 위해 [kcat](https://github.com/edenhill/kcat)을 사용합니다.
설치 유형에 맞는 Connect 설정은 [이 지침](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#set-up-a-local-connect-worker-with-cp-install)을 따르십시오. 이때 standalone과 분산 클러스터의 차이점을 유의하십시오. Confluent Cloud를 사용하는 경우에는 분산 설정이 해당됩니다.
가장 중요한 매개변수는 `http.api.url`입니다. ClickHouse의 [HTTP 인터페이스](/docs/ko/concepts/features/interfaces/http)에서는 INSERT 문을 URL 매개변수로 인코딩해야 합니다. 여기에는 포맷(이 경우 `JSONEachRow`)과 대상 데이터베이스가 포함되어야 합니다. 포맷은 Kafka 데이터와 일치해야 하며, 이 데이터는 HTTP payload에서 문자열로 변환됩니다. 이러한 매개변수는 URL 이스케이프 처리해야 합니다. GitHub 데이터셋에 대한 이 포맷의 예시는 아래와 같습니다(ClickHouse를 로컬에서 실행한다고 가정).
```response theme={null}
://:?query=INSERT%20INTO%20.%20FORMAT%20JSONEachRow
http://localhost:8123?query=INSERT%20INTO%20default.github%20FORMAT%20JSONEachRow
```
다음 추가 매개변수는 HTTP Sink를 ClickHouse와 함께 사용할 때 유용합니다. 전체 매개변수 목록은 [여기](https://docs.confluent.io/kafka-connect-http/current/connector_config.html)에서 확인할 수 있습니다.
* `request.method` - **POST**로 설정합니다.
* `retry.on.status.codes` - 모든 오류 코드에서 재시도하려면 400-500으로 설정합니다. 데이터에서 예상되는 오류에 맞게 조정하십시오.
* `request.body.format` - 대부분의 경우 JSON으로 설정합니다.
* `auth.type` - ClickHouse에서 인증을 사용하는 경우 BASIC으로 설정합니다. 현재는 ClickHouse와 호환되는 다른 인증 메커니즘은 지원되지 않습니다.
* `ssl.enabled` - SSL을 사용하는 경우 true로 설정합니다.
* `connection.user` - ClickHouse 사용자 이름입니다.
* `connection.password` - ClickHouse 비밀번호입니다.
* `batch.max.size` - 단일 배치로 전송할 행 수입니다. 충분히 큰 값으로 설정되어 있는지 확인하십시오. ClickHouse [권장 사항](/docs/ko/reference/statements/insert-into#performance-considerations)에 따르면 1000은 최소값으로 간주해야 합니다.
* `tasks.max` - HTTP Sink 커넥터는 하나 이상의 작업을 실행할 수 있습니다. 이를 통해 성능을 높일 수 있습니다. 배치 크기와 함께 성능을 개선하는 주요 수단입니다.
* `key.converter` - 키의 타입에 맞게 설정합니다.
* `value.converter` - topic의 데이터 타입에 따라 설정합니다. 이 데이터에는 스키마가 필요하지 않습니다. 여기서 사용하는 포맷은 `http.api.url` 매개변수에 지정된 FORMAT과 일치해야 합니다. 가장 간단한 방법은 JSON과 org.apache.kafka.connect.json.JsonConverter 컨버터를 사용하는 것입니다. org.apache.kafka.connect.storage.StringConverter 컨버터를 사용해 값을 문자열로 처리하는 것도 가능합니다. 다만 이 경우에는 사용자가 함수로 삽입 SQL 문에서 값을 추출해야 합니다. io.confluent.connect.avro.AvroConverter 컨버터를 사용하는 경우 ClickHouse는 [Avro 포맷](/docs/ko/reference/formats/Avro/Avro)도 지원합니다.
프록시, 재시도, 고급 SSL 구성 방법을 포함한 전체 설정 목록은 [여기](https://docs.confluent.io/kafka-connect-http/current/connector_config.html)에서 확인할 수 있습니다.
GitHub 샘플 데이터용 예시 설정 파일은 [여기](https://github.com/ClickHouse/clickhouse-docs/tree/main/docs/integrations/data-ingestion/kafka/code/connectors/http_sink)에서 확인할 수 있습니다. 이는 Connect가 standalone 모드로 실행되고 Kafka가 Confluent Cloud에서 호스팅된다고 가정합니다.
테이블이 생성되었는지 확인하십시오. 표준 MergeTree를 사용하는 최소 GitHub 데이터셋의 예시는 아래와 같습니다.
```sql theme={null}
CREATE TABLE github
(
file_time DateTime,
event_type Enum('CommitCommentEvent' = 1, 'CreateEvent' = 2, 'DeleteEvent' = 3, 'ForkEvent' = 4,'GollumEvent' = 5, 'IssueCommentEvent' = 6, 'IssuesEvent' = 7, 'MemberEvent' = 8, 'PublicEvent' = 9, 'PullRequestEvent' = 10, 'PullRequestReviewCommentEvent' = 11, 'PushEvent' = 12, 'ReleaseEvent' = 13, 'SponsorshipEvent' = 14, 'WatchEvent' = 15, 'GistEvent' = 16, 'FollowEvent' = 17, 'DownloadEvent' = 18, 'PullRequestReviewEvent' = 19, 'ForkApplyEvent' = 20, 'Event' = 21, 'TeamAddEvent' = 22),
actor_login LowCardinality(String),
repo_name LowCardinality(String),
created_at DateTime,
updated_at DateTime,
action Enum('none' = 0, 'created' = 1, 'added' = 2, 'edited' = 3, 'deleted' = 4, 'opened' = 5, 'closed' = 6, 'reopened' = 7, 'assigned' = 8, 'unassigned' = 9, 'labeled' = 10, 'unlabeled' = 11, 'review_requested' = 12, 'review_request_removed' = 13, 'synchronize' = 14, 'started' = 15, 'published' = 16, 'update' = 17, 'create' = 18, 'fork' = 19, 'merged' = 20),
comment_id UInt64,
path String,
ref LowCardinality(String),
ref_type Enum('none' = 0, 'branch' = 1, 'tag' = 2, 'repository' = 3, 'unknown' = 4),
creator_user_login LowCardinality(String),
number UInt32,
title String,
labels Array(LowCardinality(String)),
state Enum('none' = 0, 'open' = 1, 'closed' = 2),
assignee LowCardinality(String),
assignees Array(LowCardinality(String)),
closed_at DateTime,
merged_at DateTime,
merge_commit_sha String,
requested_reviewers Array(LowCardinality(String)),
merged_by LowCardinality(String),
review_comments UInt32,
member_login LowCardinality(String)
) ENGINE = MergeTree ORDER BY (event_type, repo_name, created_at)
```
Kafka에 메시지를 전송합니다. 아래에서는 [kcat](https://github.com/edenhill/kcat)를 사용해 1만 개의 메시지를 전송합니다.
```bash theme={null}
head -n 10000 github_all_columns.ndjson | kcat -b : -X security.protocol=sasl_ssl -X sasl.mechanisms=PLAIN -X sasl.username= -X sasl.password= -t github
```
대상 테이블 "Github"를 간단히 조회해 보면 데이터가 삽입된 것을 확인할 수 있습니다.
```sql theme={null}
SELECT count() FROM default.github;
| count\(\) |
| :--- |
| 10000 |
```
