Confluent HTTP Sink コネクタ

Kafka Connect と ClickHouse で HTTP Sink コネクタを利用する

​

Confluent HTTP Sink コネクタ

HTTP Sink コネクタはデータ型に依存しないため、Kafka のスキーマを必要とせず、Maps や Arrays などの ClickHouse 固有のデータ型にも対応しています。この柔軟性がある一方で、設定はやや複雑になります。

以下では、単一の Kafka トピックからメッセージを取り込み、ClickHouse テーブルに行を挿入するシンプルな導入方法について説明します。

HTTP コネクタは Confluent Enterprise License の下で配布されています。

​

クイックスタート手順

​

1. 接続情報を確認する

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

Parameter(s)	Description
HOST and PORT	通常、TLS を使用する場合のポートは 8443、TLS を使用しない場合は 8123 です。
DATABASE NAME	デフォルトでは default という名前のデータベースがあります。接続先のデータベース名を使用してください。
USERNAME and PASSWORD	デフォルトのユーザー名は default です。用途に応じたユーザー名を使用してください。

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

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

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

​

2. Kafka Connect と HTTP Sink コネクタを実行する

選択肢は 2 つあります。

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

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

以下の例では Confluent Cloud を使用します。

​

3. ClickHouse に宛先テーブルを作成する

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

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

​

4. HTTP Sink の設定

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 のパスワード

この 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) で適切ですが、より堅牢に再試行したい場合は調整してもかまいません。

​

5. 接続のテスト

HTTP Sink で設定したトピックにメッセージを作成します

作成したメッセージが ClickHouse インスタンスに書き込まれていることを確認します。

​

トラブルシューティング

​

HTTP Sink がメッセージをバッチ化しない

Sink のドキュメントより:

Kafka ヘッダー値が異なるメッセージを含む場合、HTTP Sink コネクタはリクエストをバッチ化しません。

1. Kafka レコードのキーが同じであることを確認してください。
2. HTTP API の URL にパラメータを追加すると、レコードごとに一意の URL になることがあります。そのため、追加の URL パラメータを使用するとバッチ化は無効になります。

​

400 Bad Request

CANNOT_PARSE_QUOTED_STRING

JSONオブジェクトを String カラムに挿入する際に、HTTP Sink が次のメッセージを出して失敗する場合:

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 を URL エンコードされた文字列 SETTINGS%20input_format_json_read_objects_as_strings%3D1 として指定します

​

GitHub データセットを読み込む (任意)

この例では、GitHub データセットの Array フィールドを保持したまま扱います。サンプルでは、空の github トピックがあることを前提とし、Kafka へのメッセージの挿入に kcat を使用します。

1. 設定を準備する

インストール形態に応じた 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 でホストされていることを前提としています。

2. ClickHouseテーブルを作成する

テーブルが作成済みであることを確認してください。標準的な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)

3. Kafka にデータを追加する

Kafka にメッセージを投入します。以下では、kcat を使用して 1 万件のメッセージを投入します。

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\(\) |
| :--- |
| 10000 |