ClickHouse/docs/ja/integrations/data-ingestion/kafka/kafka-vector.md

sidebar_label	sidebar_position	slug	description
KafkaとのVector	3	/ja/integrations/kafka/kafka-vector	KafkaとClickHouseでVectorを使用する

import ConnectionDetails from '@site/docs/ja/_snippets/_gather_your_details_http.mdx';

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形式または文字列としてKafkaに公開される必要があります。

ライセンス

Vectorは、MPL-2.0ライセンスの下で配布されています。

接続情報を収集する

手順

1. 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行で構成されています。

2. ターゲットテーブルが作成されていることを確認してください。以下ではデフォルトのデータベースを使用します。

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をダウンロードしてインストールします。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シンクに渡されることを保証します。文字列としてメッセージを扱い、デフォルトのbytes値を使用する場合、メッセージの内容はmessageフィールドに追加されます。ほとんどの場合、これはVector入門ガイドで説明されているようにClickHouseで処理が必要です。
• Vectorはメッセージに多くのフィールドを追加します。この例では、ClickHouseシンクの設定パラメータskip_unknown_fields = trueを介してこれらのフィールドを無視しています。これは、ターゲットテーブルスキーマの一部でないフィールドを無視します。例えばoffsetなどのメタフィールドが追加されるようにスキーマを調整しても問題ありません。
• シンクがソースイベントへの参照としてinputsパラメータをどのように使用するかに注意してください。
• ClickHouseシンクの動作についてはこちらに記載されています。最適なスループットを得るために、ユーザーはbuffer.max_events、batch.timeout_secsおよびbatch.max_bytesパラメータのチューニングを行うことができます。 ClickHouseの推奨事項に従い、1000のイベント数は1バッチあたりの最低値と考えられるべきです。高スループットのユースケースでは、buffer.max_events パラメータを増やすことができます。スループットに変動がある場合は、batch.timeout_secs パラメータを変更する必要があるかもしれません。
• auto_offset_reset = "smallest"パラメータは、Kafkaソースがトピックの先頭から開始することを強制します。これにより、ステップ(1)で公開されたメッセージを消費することが保証されます。異なる動作を必要とする場合はこちらをご覧ください。

1. Vectorを開始します。

vector --config ./kafka.toml

デフォルトでは、ClickHouseへの挿入が開始される前にヘルスチェックが必要です。これは接続可能性を確保し、スキーマを確認するためです。問題が発生した場合にさらに詳細なログを取得するには、VECTOR_LOG=debugを付けることが役立つ場合があります。

5. データの挿入を確認します。

SELECT count() as count FROM github;

count
200000