Kafka 및 ClickHouse와 함께 Vector 사용하기

Vector는 Kafka에서 데이터를 읽고 이벤트를 ClickHouse로 전송할 수 있는 벤더 중립적인 데이터 파이프라인입니다. ClickHouse와 함께 Vector를 사용하는 시작하기 가이드는 로그 사용 사례와 파일에서 이벤트를 읽는 방법에 중점을 둡니다. 여기서는 Kafka 토픽에 저장된 이벤트가 포함된 GitHub 샘플 데이터셋을 사용합니다. Vector는 푸시 또는 풀 모델을 통해 데이터를 가져오기 위해 sources를 사용합니다. 반면 Sinks는 이벤트의 대상을 제공합니다. 따라서 여기서는 Kafka 소스와 ClickHouse 싱크를 사용합니다. Kafka는 싱크로 지원되지만 ClickHouse 소스는 제공되지 않습니다. 따라서 ClickHouse에서 Kafka로 데이터를 전송하려는 경우 Vector는 적합하지 않습니다. Vector는 데이터 변환도 지원합니다. 하지만 이는 이 가이드의 범위를 벗어납니다. 데이터셋에 이 기능이 필요하면 Vector 문서를 참조하십시오. 현재 ClickHouse 싱크 구현은 HTTP 인터페이스를 사용합니다. 현재 ClickHouse 싱크는 JSON 스키마 사용을 지원하지 않습니다. 데이터는 일반 JSON 포맷 또는 String 형식으로 Kafka에 게시되어야 합니다.

라이선스

Vector는 MPL-2.0 라이선스로 배포됩니다

connection 세부 정보를 확인합니다

HTTP(S)로 ClickHouse에 연결하려면 다음 정보가 필요합니다.

매개변수	설명
`HOST` and `PORT`	일반적으로 TLS를 사용하는 경우 포트는 8443, TLS를 사용하지 않는 경우 8123입니다.
`DATABASE NAME`	기본적으로 `default`라는 이름의 데이터베이스가 제공되며, 연결할 데이터베이스 이름을 사용하십시오.
`USERNAME` and `PASSWORD`	기본 사용자 이름은 `default`입니다. 사용 사례에 맞는 사용자 이름을 사용하십시오.

ClickHouse Cloud 서비스의 연결 정보는 ClickHouse Cloud 콘솔에서 확인할 수 있습니다. 서비스를 선택한 다음 Connect를 클릭하십시오.

HTTPS를 선택하십시오. 연결 정보가 예시 curl 명령으로 표시됩니다.

자가 관리형 ClickHouse를 사용하는 경우 연결 정보는 ClickHouse 관리자가 설정합니다.

단계

Kafka github 토픽을 생성한 다음 Github 데이터셋을 적재합니다.

cat /opt/data/github/github_all_columns.ndjson | kcat -b <host>:<port> -X security.protocol=sasl_ssl -X sasl.mechanisms=PLAIN -X sasl.username=<username> -X sasl.password=<password> -t github

이 데이터셋은 ClickHouse/ClickHouse 리포지토리를 중심으로 한 200,000개 행으로 이루어져 있습니다.

대상 테이블(target table)이 생성되었는지 확인하세요. 아래에서는 기본 데이터베이스를 사용합니다.

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);

Vector를 다운로드하여 설치합니다. kafka.toml 설정 파일을 만들고 Kafka 및 ClickHouse 인스턴스에 맞게 값을 수정합니다.

[sources.github]
type = "kafka"
auto_offset_reset = "smallest"
bootstrap_servers = "<kafka_host>:<kafka_port>"
group_id = "vector"
topics = [ "github" ]
tls.enabled = true
sasl.enabled = true
sasl.mechanism = "PLAIN"
sasl.username = "<username>"
sasl.password = "<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 시작하기 가이드에 설명된 대로 ClickHouse에서 처리해야 합니다.
Vector는 메시지에 여러 필드를 추가합니다. 이 예시에서는 구성 매개변수 skip_unknown_fields = true를 사용해 ClickHouse 싱크에서 이러한 필드를 무시합니다. 이렇게 하면 대상 테이블 스키마에 포함되지 않은 필드가 무시됩니다. 필요에 따라 스키마를 조정하여 offset과 같은 메타 필드가 포함되도록 하십시오.
inputs 매개변수를 통해 싱크가 이벤트 소스를 참조하는 방식에 유의하십시오.
여기에 설명된 ClickHouse 싱크의 동작에 유의하십시오. 최적의 처리량을 위해 buffer.max_events, batch.timeout_secs, batch.max_bytes 매개변수를 조정하는 것이 좋습니다. ClickHouse 권장 사항에 따르면 단일 batch의 이벤트 수는 최소 1000개로 설정하는 것이 좋습니다. 일정한 고처리량 사용 사례에서는 buffer.max_events 매개변수를 늘릴 수 있습니다. 처리량 변동이 큰 경우에는 batch.timeout_secs 매개변수를 조정해야 할 수 있습니다.
매개변수 auto_offset_reset = "smallest"는 Kafka 소스가 토픽의 처음부터 시작하도록 강제하므로 단계 (1)에서 게시한 메시지를 소비하게 됩니다. 필요한 동작은 다를 수 있습니다. 자세한 내용은 여기를 참조하십시오.

Vector 시작

vector --config ./kafka.toml

기본적으로 ClickHouse에 대한 데이터 삽입이 시작되기 전에 헬스 체크가 필요합니다. 이를 통해 연결이 가능한지, 스키마(schema)를 읽을 수 있는지 확인합니다. 문제가 발생한 경우 진단에 도움이 되는 추가 로깅을 확인하려면 VECTOR_LOG=debug를 앞에 붙이십시오.

데이터가 삽입되었는지 확인합니다.

SELECT count() AS count FROM github;

개수
200000