%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) で適切ですが、より堅牢に再試行したい場合は調整してもかまいません。
HTTP Sink で設定したトピックにメッセージを作成します
続いて、作成したメッセージが ClickHouse インスタンスに書き込まれていることを確認します。
### トラブルシューティング
#### HTTP Sink がメッセージをバッチ化しない
[Sink のドキュメント](https://docs.confluent.io/kafka-connectors/http/current/overview.html#http-sink-connector-for-cp)より:
> Kafka ヘッダー値が異なるメッセージを含む場合、HTTP Sink コネクタはリクエストをバッチ化しません。
1. Kafka レコードのキーが同じであることを確認してください。
2. HTTP API の URL にパラメータを追加すると、レコードごとに一意の URL になることがあります。そのため、追加の URL パラメータを使用するとバッチ化は無効になります。
#### 400 Bad Request
##### CANNOT\_PARSE\_QUOTED\_STRING
JSONオブジェクトを `String` カラムに挿入する際に、HTTP Sink が次のメッセージを出して失敗する場合:
```response theme={null}
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](https://github.com/edenhill/kcat) を使用します。
インストール形態に応じた Connect のセットアップについては、スタンドアロンと分散クラスターの違いに注意しつつ、[こちらの手順](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#set-up-a-local-connect-worker-with-cp-install)に従ってください。Confluent Cloud を使用する場合は、分散セットアップが該当します。
最も重要なパラメータは `http.api.url` です。ClickHouse の [HTTP インターフェイス](/docs/ja/concepts/features/interfaces/http) では、INSERT ステートメントを URL のパラメータとしてエンコードする必要があります。これには、フォーマット (この場合は `JSONEachRow`) と移行先データベースを含める必要があります。フォーマットは Kafka のデータと一致している必要があり、そのデータは HTTP ペイロード内で文字列に変換されます。これらのパラメータは URL エスケープする必要があります。GitHub データセットに対するこのフォーマットの例 (ClickHouse をローカルで実行していることを前提) は、以下のとおりです。
```response theme={null}
://:?query=INSERT%20INTO%20.%20FORMAT%20JSONEachRow
http://localhost:8123?query=INSERT%20INTO%20default.github%20FORMAT%20JSONEachRow
```
ClickHouse で HTTP Sink を使用する際は、以下の追加パラメータが関係します。完全なパラメータ一覧は[こちら](https://docs.confluent.io/kafka-connect-http/current/connector_config.html)で確認できます。
* `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 の[推奨事項](/docs/ja/reference/statements/insert-into#performance-considerations)によると、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](/docs/ja/reference/formats/Avro/Avro) もサポートしています。
proxy、再試行、高度な SSL の設定方法を含む設定の完全な一覧は、[こちら](https://docs.confluent.io/kafka-connect-http/current/connector_config.html)で確認できます。
GitHub サンプルデータ用の設定ファイル例は[こちら](https://github.com/ClickHouse/clickhouse-docs/tree/main/docs/integrations/data-ingestion/kafka/code/connectors/http_sink)にあります。これは、Connect がスタンドアロン モードで実行され、Kafka が Confluent Cloud でホストされていることを前提としています。
テーブルが作成されていることを確認してください。標準的な MergeTree を使用する最小構成の GitHub データセットの例を以下に示します。
```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)
```
Kafka にメッセージを書き込みます。以下では、[kcat](https://github.com/edenhill/kcat) を使って 10k 件のメッセージを書き込みます。
```bash theme={null}
head -n 10000 github_all_columns.ndjson | kcat -b : -X security.protocol=sasl_ssl -X sasl.mechanisms=PLAIN -X sasl.username= -X sasl.password= -t github
```
ターゲットテーブル "Github" を簡単に参照すれば、データが挿入されたことを確認できます。
```sql theme={null}
SELECT count() FROM default.github;
| count\(\) |
| :--- |
| 10000 |
```
