> ## Documentation Index
> Fetch the complete documentation index at: https://private-7c7dfe99-home-button.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.
> Использование коннектора HTTP Sink с Kafka Connect и ClickHouse
# Confluent коннектор HTTP Sink
export const Image = ({img, alt, size}) => {
return
;
};
Коннектор HTTP Sink не зависит от типа данных, поэтому ему не требуется схема Kafka; кроме того, он поддерживает специфичные для ClickHouse типы данных, такие как Map и Array. За эту дополнительную гибкость приходится платить небольшим усложнением конфигурации.
Ниже мы опишем простую установку: получение сообщений из одного топика Kafka и вставку строк в таблицу ClickHouse.
Коннектор HTTP Sink распространяется по лицензии [Confluent Enterprise License](https://docs.confluent.io/kafka-connect-http/current/overview.html#license).
### Шаги быстрого старта
#### 1. Подготовьте сведения о подключении
Чтобы подключиться к ClickHouse по HTTP(S), вам понадобится следующая информация:
| Параметр(ы) | Описание |
| ------------------------- | ---------------------------------------------------------------------------------------------------------------------- |
| `HOST` and `PORT` | Обычно используется порт 8443 при использовании TLS и 8123 без TLS. |
| `DATABASE NAME` | По умолчанию есть база данных `default`; используйте имя базы данных, к которой хотите подключиться. |
| `USERNAME` and `PASSWORD` | По умолчанию имя пользователя — `default`. Используйте имя пользователя, подходящее для вашего сценария использования. |
Сведения о подключении для вашего сервиса ClickHouse Cloud доступны в консоли ClickHouse Cloud.
Выберите сервис и нажмите **Connect**:
Выберите **HTTPS**. Сведения о подключении будут показаны в примере команды `curl`.
Если вы используете самоуправляемый ClickHouse, сведения о подключении задаёт ваш администратор ClickHouse.
#### 2. Запустите Kafka Connect и коннектор HTTP Sink
У вас есть два варианта:
* **Самоуправляемый:** Скачайте пакет Confluent и установите его локально. Следуйте инструкциям по установке коннектора, приведённым [здесь](https://docs.confluent.io/kafka-connect-http/current/overview.html).
Если вы используете метод установки через confluent-hub, локальные файлы конфигурации будут обновлены.
* **Confluent Cloud:** Для пользователей Confluent Cloud в качестве хостинга Kafka доступен полностью управляемый HTTP Sink. Для этого среда ClickHouse должна быть доступна из Confluent Cloud.
В следующих примерах используется Confluent Cloud.
#### 3. Создайте целевую таблицу в ClickHouse
Перед проверкой подключения сначала создадим тестовую таблицу в ClickHouse Cloud — в неё будут поступать данные из Kafka:
```sql theme={null}
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` - URL ClickHouse Cloud с указанным запросом `INSERT` `://:?query=INSERT%20INTO%20.%20FORMAT%20JSONEachRow`. **Примечание**: запрос должен быть URL-кодирован.
* `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 proxy.
* `Maximum Reties` - значение по умолчанию (10) подходит, но при необходимости вы можете увеличить его для более надёжных повторных попыток.
#### 5. Проверка подключения
Создайте сообщение в топике, настроенном для вашего HTTP Sink
и убедитесь, что созданное сообщение появилось в вашем экземпляре ClickHouse.
### Устранение неполадок
#### HTTP Sink не выполняет батчинг сообщений
Из [документации по Sink](https://docs.confluent.io/kafka-connectors/http/current/overview.html#http-sink-connector-for-cp):
> Коннектор HTTP Sink не выполняет батчинг запросов для сообщений с разными значениями заголовков Kafka.
1. Убедитесь, что у ваших записей Kafka одинаковый ключ.
2. При добавлении параметров в URL HTTP API для каждой записи может формироваться уникальный URL. По этой причине батчинг отключается при использовании дополнительных параметров URL.
#### 400 Некорректный запрос
##### 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)
```
Установите параметр `input_format_json_read_objects_as_strings=1` в URL в виде закодированной строки `SETTINGS%20input_format_json_read_objects_as_strings%3D1`
### Загрузите набор данных GitHub (необязательно)
Обратите внимание: в этом примере сохраняются поля типа Array из набора данных GitHub. Мы предполагаем, что в примерах у вас есть пустой топик github, и для вставки сообщений в Kafka используем [kcat](https://github.com/edenhill/kcat).
##### 1. Подготовьте конфигурацию
Следуйте [этим инструкциям](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#set-up-a-local-connect-worker-with-cp-install) по настройке Connect в соответствии с типом вашей установки, учитывая различия между автономным и распределённым кластером. Если вы используете Confluent Cloud, вам нужен распределённый вариант.
Самый важный параметр — `http.api.url`. [HTTP-интерфейс](/ru/concepts/features/interfaces/http) ClickHouse требует передавать оператор 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
```
Следующие дополнительные параметры относятся к использованию коннектора HTTP Sink с ClickHouse. Полный список параметров можно найти [здесь](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` - Установите значение BASIC, если для ClickHouse используется BASIC-аутентификация. Другие совместимые с ClickHouse механизмы аутентификации в настоящее время не поддерживаются.
* `ssl.enabled` - установите значение true, если используете SSL.
* `connection.user` - имя пользователя ClickHouse.
* `connection.password` - пароль ClickHouse.
* `batch.max.size` - Количество строк, отправляемых в одном батче. Убедитесь, что здесь задано достаточно большое значение. Согласно [рекомендациям](/ru/reference/statements/insert-into#performance-considerations) ClickHouse, значение 1000 следует считать минимальным.
* `tasks.max` - Коннектор HTTP Sink поддерживает запуск одной или нескольких задач. Это можно использовать для повышения производительности. Наряду с размером батча это основной способ повысить производительность.
* `key.converter` - задайте в соответствии с типами ваших ключей.
* `value.converter` - задайте в зависимости от типа данных в вашем топике. Для этих данных схема не требуется. Формат здесь должен соответствовать FORMAT, указанному в параметре `http.api.url`. Проще всего использовать JSON и конвертер org.apache.kafka.connect.json.JsonConverter. Также можно обрабатывать значение как строку с помощью конвертера org.apache.kafka.connect.storage.StringConverter, хотя в этом случае пользователю потребуется извлекать значение в операторе вставки с помощью функций. [Формат Avro](/ru/reference/formats/Avro/Avro) также поддерживается в ClickHouse при использовании конвертера io.confluent.connect.avro.AvroConverter.
Полный список настроек, включая сведения о том, как настроить прокси, повторные попытки и расширенные параметры 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.
##### 2. Создайте таблицу в ClickHouse
Убедитесь, что таблица создана. Ниже приведён пример минимального набора данных GitHub с использованием стандартного движка MergeTree.
```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)
```
##### 3. Добавьте данные в Kafka
Отправьте сообщения в Kafka. Ниже мы используем [kcat](https://github.com/edenhill/kcat), чтобы отправить 10 тыс. сообщений.
```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 |
```