Confluent HTTP Sink コネクタ

HTTP(S) で ClickHouse に接続するには、次の情報が必要です。

ClickHouse Cloud サービスの詳細は、ClickHouse Cloud コンソールで確認できます。 サービスを選択し、Connect をクリックします。

HTTPS を選択します。接続情報は curl コマンドの例として表示されます。

セルフマネージド ClickHouse を使用している場合、接続情報は ClickHouse 管理者によって設定されます。

2 つの選択肢があります。

• セルフマネージド: Confluent パッケージをダウンロードしてローカルにインストールします。コネクタのインストールについては、こちらに記載されている手順に従ってください。 confluent-hub によるインストール方法を使用する場合、ローカルの設定ファイルが更新されます。

• Confluent Cloud: Kafka のホスティングに Confluent Cloud を使用している場合は、完全マネージド型の HTTP Sink を利用できます。この場合、ClickHouse 環境が Confluent Cloud からアクセス可能である必要があります。

接続テストの前に、まず ClickHouse Cloud にテスト用のテーブルを作成します。このテーブルが Kafka からのデータを受信します。

CREATE TABLE default.my_table
(
    `side` String,
    `quantity` Int32,
    `symbol` String,
    `price` Int32,
    `account` String,
    `userid` String
)
ORDER BY tuple()

Kafka トピックと HTTP Sink コネクタのインスタンスを作成します。

HTTP Sink コネクタを設定します。

• 作成したトピック名を指定します
• 認証
  • HTTP Url - INSERT クエリを指定した ClickHouse Cloud の URL <protocol>://<clickhouse_host>:<clickhouse_port>?query=INSERT%20INTO%20<database>.<table>%20FORMAT%20JSONEachRow。注: クエリはエンコードする必要があります。
  • Endpoint Authentication type - BASIC
  • Auth username - ClickHouse のユーザー名
  • Auth password - ClickHouse のパスワード

• 設定
  • 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 インスタンスに書き込まれていることを確認します。

インストール形態に応じた Connect のセットアップについては、スタンドアロンと分散クラスターの違いに注意しつつ、こちらの手順に従ってください。Confluent Cloud を使用する場合は、分散セットアップが該当します。

最も重要なパラメータは http.api.url です。ClickHouse の HTTP インターフェイス では、INSERT ステートメントを URL のパラメータとしてエンコードする必要があります。これには、フォーマット (この場合は JSONEachRow) と移行先データベースを含める必要があります。フォーマットは Kafka のデータと一致している必要があり、そのデータは HTTP ペイロード内で文字列に変換されます。これらのパラメータは URL エスケープする必要があります。GitHub データセットに対するこのフォーマットの例 (ClickHouse をローカルで実行していることを前提) は、以下のとおりです。

<protocol>://<clickhouse_host>:<clickhouse_port>?query=INSERT%20INTO%20<database>.<table>%20FORMAT%20JSONEachRow

http://localhost:8123?query=INSERT%20INTO%20default.github%20FORMAT%20JSONEachRow

ClickHouse で HTTP Sink を使用する際は、以下の追加パラメータが関係します。完全なパラメータ一覧はこちらで確認できます。

• 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 - 1 回の batch で送信する行数です。十分に大きな値を設定してください。ClickHouse の推奨事項によると、1000 は最低値と考えるべきです。
• tasks.max - HTTP Sink コネクタは 1 つ以上のタスクの実行をサポートしています。これはパフォーマンス向上に利用できます。batch size とあわせて、パフォーマンス改善の主要な手段となります。
• key.converter - キーの型に応じて設定します。
• value.converter - topic 上のデータ型に基づいて設定します。このデータにスキーマは不要です。ここでのフォーマットは、パラメータ http.api.url で指定する FORMAT と一致している必要があります。最も簡単なのは、JSON と org.apache.kafka.connect.json.JsonConverter コンバータを使用する方法です。org.apache.kafka.connect.storage.StringConverter コンバータを使って値を文字列として扱うことも可能ですが、その場合は INSERT ステートメント内で関数を使って値を抽出する必要があります。io.confluent.connect.avro.AvroConverter コンバータを使用する場合、ClickHouse は Avro format もサポートしています。

proxy、再試行、高度な SSL の設定方法を含む設定の完全な一覧は、こちらで確認できます。

GitHub サンプルデータ用の設定ファイル例はこちらにあります。これは、Connect がスタンドアロン モードで実行され、Kafka が Confluent Cloud でホストされていることを前提としています。

テーブルが作成されていることを確認してください。標準的な MergeTree を使用する最小構成の GitHub データセットの例を以下に示します。

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 を使って 10k 件のメッセージを書き込みます。

head -n 10000 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

ターゲットテーブル "Github" を簡単に参照すれば、データが挿入されたことを確認できます。

SELECT count() FROM default.github;

| count\