> ## Documentation Index > Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt > Use this file to discover all available pages before exploring further. > Kafka 및 ClickHouse와 함께 Vector 사용하기 # Kafka 및 ClickHouse와 함께 Vector 사용하기
## Kafka 및 ClickHouse와 함께 Vector 사용하기
Vector는 Kafka에서 데이터를 읽고 이벤트를 ClickHouse로 전송할 수 있는 벤더 중립적인 데이터 파이프라인입니다. ClickHouse와 함께 Vector를 사용하는 [시작하기](/docs/ko/integrations/connectors/data-ingestion/etl-tools/vector-to-clickhouse) 가이드는 로그 사용 사례와 파일에서 이벤트를 읽는 방법에 중점을 둡니다. 여기서는 Kafka 토픽에 저장된 이벤트가 포함된 [GitHub 샘플 데이터셋](https://datasets-documentation.s3.eu-west-3.amazonaws.com/kafka/github_all_columns.ndjson)을 사용합니다. Vector는 푸시 또는 풀 모델을 통해 데이터를 가져오기 위해 [sources](https://vector.dev/docs/introduction/concepts/#sources)를 사용합니다. 반면 [Sinks](https://vector.dev/docs/introduction/concepts/#sinks)는 이벤트의 대상을 제공합니다. 따라서 여기서는 Kafka 소스와 ClickHouse 싱크를 사용합니다. Kafka는 싱크로 지원되지만 ClickHouse 소스는 제공되지 않습니다. 따라서 ClickHouse에서 Kafka로 데이터를 전송하려는 경우 Vector는 적합하지 않습니다. Vector는 데이터 [변환](https://vector.dev/docs/reference/configuration/transforms/)도 지원합니다. 하지만 이는 이 가이드의 범위를 벗어납니다. 데이터셋에 이 기능이 필요하면 Vector 문서를 참조하십시오. 현재 ClickHouse 싱크 구현은 HTTP 인터페이스를 사용합니다. 현재 ClickHouse 싱크는 JSON 스키마 사용을 지원하지 않습니다. 데이터는 일반 JSON 포맷 또는 String 형식으로 Kafka에 게시되어야 합니다.
### 라이선스
Vector는 [MPL-2.0 라이선스](https://github.com/vectordotdev/vector/blob/master/LICENSE)로 배포됩니다
### connection 세부 정보를 확인합니다
HTTP(S)로 ClickHouse에 연결하려면 다음 정보가 필요합니다. | 매개변수 | 설명 | | ------------------------- | ---------------------------------------------------------- | | `HOST` and `PORT` | 일반적으로 TLS를 사용하는 경우 포트는 8443, TLS를 사용하지 않는 경우 8123입니다. | | `DATABASE NAME` | 기본적으로 `default`라는 이름의 데이터베이스가 제공되며, 연결할 데이터베이스 이름을 사용하십시오. | | `USERNAME` and `PASSWORD` | 기본 사용자 이름은 `default`입니다. 사용 사례에 맞는 사용자 이름을 사용하십시오. | ClickHouse Cloud 서비스의 연결 정보는 ClickHouse Cloud 콘솔에서 확인할 수 있습니다. 서비스를 선택한 다음 **Connect**를 클릭하십시오.
ClickHouse Cloud 서비스 연결 버튼
**HTTPS**를 선택하십시오. 연결 정보가 예시 `curl` 명령으로 표시됩니다.
ClickHouse Cloud HTTPS 연결 정보
자가 관리형 ClickHouse를 사용하는 경우 연결 정보는 ClickHouse 관리자가 설정합니다.
### 단계
1. Kafka `github` 토픽을 생성한 다음 [Github 데이터셋](https://datasets-documentation.s3.eu-west-3.amazonaws.com/kafka/github_all_columns.ndjson)을 적재합니다. ```bash theme={null} cat /opt/data/github/github_all_columns.ndjson | kcat -b : -X security.protocol=sasl_ssl -X sasl.mechanisms=PLAIN -X sasl.username= -X sasl.password= -t github ``` 이 데이터셋은 `ClickHouse/ClickHouse` 리포지토리를 중심으로 한 200,000개 행으로 이루어져 있습니다. 2. 대상 테이블(target table)이 생성되었는지 확인하세요. 아래에서는 기본 데이터베이스를 사용합니다. ```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); ``` 3. [Vector를 다운로드하여 설치합니다](https://vector.dev/docs/setup/quickstart/). `kafka.toml` 설정 파일을 만들고 Kafka 및 ClickHouse 인스턴스에 맞게 값을 수정합니다. ```toml theme={null} [sources.github] type = "kafka" auto_offset_reset = "smallest" bootstrap_servers = ":" group_id = "vector" topics = [ "github" ] tls.enabled = true sasl.enabled = true sasl.mechanism = "PLAIN" sasl.username = "" sasl.password = "" decoding.codec = "json" [sinks.clickhouse] type = "clickhouse" inputs = ["github"] endpoint = "http://localhost:8123" database = "default" table = "github" skip_unknown_fields = true auth.strategy = "basic" auth.user = "username" auth.password = "password" buffer.max_events = 10000 batch.timeout_secs = 1 ``` 이 구성과 Vector의 동작에 대해 몇 가지 중요한 참고 사항이 있습니다. * 이 예시는 Confluent Cloud를 기준으로 테스트되었습니다. 따라서 `sasl.*` 및 `ssl.enabled` 보안 옵션은 자가 관리형 환경에는 적합하지 않을 수 있습니다. * 구성 매개변수 `bootstrap_servers`에는 프로토콜 접두사가 필요하지 않습니다. 예: `pkc-2396y.us-east-1.aws.confluent.cloud:9092` * 소스 매개변수 `decoding.codec = "json"`은 메시지가 단일 JSON 객체로 ClickHouse 싱크에 전달되도록 합니다. 메시지를 `String`으로 처리하고 기본값인 `bytes`를 사용하는 경우에는 메시지 내용이 `message` 필드에 추가됩니다. 대부분의 경우 이는 [Vector 시작하기](/docs/ko/integrations/connectors/data-ingestion/etl-tools/vector-to-clickhouse#4-parse-the-logs) 가이드에 설명된 대로 ClickHouse에서 처리해야 합니다. * Vector는 메시지에 [여러 필드를 추가합니다](https://vector.dev/docs/reference/configuration/sources/kafka/#output-data). 이 예시에서는 구성 매개변수 `skip_unknown_fields = true`를 사용해 ClickHouse 싱크에서 이러한 필드를 무시합니다. 이렇게 하면 대상 테이블 스키마에 포함되지 않은 필드가 무시됩니다. 필요에 따라 스키마를 조정하여 `offset`과 같은 메타 필드가 포함되도록 하십시오. * `inputs` 매개변수를 통해 싱크가 이벤트 소스를 참조하는 방식에 유의하십시오. * [여기](https://vector.dev/docs/reference/configuration/sinks/clickhouse/#buffers-and-batches)에 설명된 ClickHouse 싱크의 동작에 유의하십시오. 최적의 처리량을 위해 `buffer.max_events`, `batch.timeout_secs`, `batch.max_bytes` 매개변수를 조정하는 것이 좋습니다. ClickHouse [권장 사항](/docs/ko/reference/statements/insert-into#performance-considerations)에 따르면 단일 batch의 이벤트 수는 최소 1000개로 설정하는 것이 좋습니다. 일정한 고처리량 사용 사례에서는 `buffer.max_events` 매개변수를 늘릴 수 있습니다. 처리량 변동이 큰 경우에는 `batch.timeout_secs` 매개변수를 조정해야 할 수 있습니다. * 매개변수 `auto_offset_reset = "smallest"`는 Kafka 소스가 토픽의 처음부터 시작하도록 강제하므로 단계 (1)에서 게시한 메시지를 소비하게 됩니다. 필요한 동작은 다를 수 있습니다. 자세한 내용은 [여기](https://vector.dev/docs/reference/configuration/sources/kafka/#auto_offset_reset)를 참조하십시오. 4. Vector 시작 ```bash theme={null} vector --config ./kafka.toml ``` 기본적으로 ClickHouse에 대한 데이터 삽입이 시작되기 전에 [헬스 체크](https://vector.dev/docs/reference/configuration/sinks/clickhouse/#healthcheck)가 필요합니다. 이를 통해 연결이 가능한지, 스키마(schema)를 읽을 수 있는지 확인합니다. 문제가 발생한 경우 진단에 도움이 되는 추가 로깅을 확인하려면 `VECTOR_LOG=debug`를 앞에 붙이십시오. 5. 데이터가 삽입되었는지 확인합니다. ```sql theme={null} SELECT count() AS count FROM github; ``` | 개수 | | :----- | | 200000 |