%20FORMAT%20JSONEachRow`. **Nota**: la consulta debe estar codificada.
* `Endpoint Authentication type` - BASIC
* `Auth username` - nombre de usuario de ClickHouse
* `Auth password` - contraseña de ClickHouse
Esta HTTP Url es propensa a errores. Asegúrese de que el escaping sea preciso para evitar problemas.
* Configuración
* `Input Kafka record value format` Depende de los datos de origen, pero en la mayoría de los casos será JSON o Avro. En la siguiente configuración asumimos `JSON`.
* En la sección `advanced configurations`:
* `HTTP Request Method` - Establézcalo en POST
* `Request Body Format` - json
* `Batch batch size` - Según las recomendaciones de ClickHouse, establézcalo en **al menos 1000**.
* `Batch json as array` - true
* `Retry on HTTP codes` - 400-500, pero adáptelo según sea necesario; por ejemplo, esto puede cambiar si tiene un proxy HTTP delante de ClickHouse.
* `Maximum Reties` - el valor predeterminado (10) es adecuado, pero no dude en ajustarlo para lograr reintentos más robustos.
Cree un mensaje en un topic configurado por su HTTP Sink
y verifique que el mensaje creado se haya escrito en su instancia de ClickHouse.
### Solución de problemas
#### HTTP Sink no agrupa los mensajes en lotes
Según la [documentación del sink](https://docs.confluent.io/kafka-connectors/http/current/overview.html#http-sink-connector-for-cp):
> El conector HTTP Sink no agrupa solicitudes de mensajes que contienen valores de cabecera de Kafka diferentes.
1. Verifique que sus registros de Kafka tengan la misma clave.
2. Cuando agrega parámetros a la URL de la API HTTP, cada registro puede generar una URL única. Por este motivo, la agrupación en lotes se desactiva cuando se usan parámetros de URL adicionales.
#### 400 solicitud incorrecta
##### CANNOT\_PARSE\_QUOTED\_STRING
Si HTTP Sink falla y muestra el siguiente mensaje al insertar un objeto JSON en una columna `String`:
```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)
```
Defina la configuración `input_format_json_read_objects_as_strings=1` en la URL como una cadena codificada: `SETTINGS%20input_format_json_read_objects_as_strings%3D1`
### Cargar el conjunto de datos de GitHub (opcional)
Tenga en cuenta que este ejemplo conserva los campos Array del conjunto de datos de GitHub. Suponemos que dispone de un topic `github` vacío para este ejemplo y que usa [kcat](https://github.com/edenhill/kcat) para insertar mensajes en Kafka.
Siga [estas instrucciones](https://docs.confluent.io/cloud/current/cp-component/connect-cloud-config.html#set-up-a-local-connect-worker-with-cp-install) para configurar Connect según su tipo de instalación, teniendo en cuenta las diferencias entre un clúster standalone y uno distribuido. Si usa Confluent Cloud, la configuración distribuida es la adecuada.
El parámetro más importante es `http.api.url`. La [interfaz HTTP](/docs/es/concepts/features/interfaces/http) de ClickHouse requiere codificar la sentencia INSERT como un parámetro en la URL. Debe incluir el formato (`JSONEachRow` en este caso) y la base de datos de destino. El formato debe ser coherente con los datos de Kafka, que se convertirán en una cadena en el payload HTTP. Estos parámetros deben codificarse para URL. A continuación se muestra un ejemplo de este formato para el conjunto de datos de GitHub (suponiendo que está ejecutando ClickHouse localmente):
```response theme={null}
://:?query=INSERT%20INTO%20.%20FORMAT%20JSONEachRow
http://localhost:8123?query=INSERT%20INTO%20default.github%20FORMAT%20JSONEachRow
```
Los siguientes parámetros adicionales son relevantes para usar HTTP Sink con ClickHouse. Puede encontrar una lista completa de parámetros [aquí](https://docs.confluent.io/kafka-connect-http/current/connector_config.html):
* `request.method` - Establézcalo en **POST**
* `retry.on.status.codes` - Establézcalo en 400-500 para reintentar ante cualquier código de error. Ajústelo según los errores esperados en los datos.
* `request.body.format` - En la mayoría de los casos, será JSON.
* `auth.type` - Establézcalo en BASIC si utiliza autenticación con ClickHouse. Actualmente no se admiten otros mecanismos de autenticación compatibles con ClickHouse.
* `ssl.enabled` - establézcalo en true si usa SSL.
* `connection.user` - nombre de usuario de ClickHouse.
* `connection.password` - contraseña de ClickHouse.
* `batch.max.size` - El número de filas que se enviarán en un solo lote. Asegúrese de que este valor sea lo suficientemente grande. Según las [recomendaciones](/docs/es/reference/statements/insert-into#performance-considerations) de ClickHouse, un valor de 1000 debe considerarse el mínimo.
* `tasks.max` - El conector HTTP Sink admite ejecutar una o más tareas. Esto puede usarse para aumentar el rendimiento. Junto con el tamaño del lote, este es el principal medio para mejorar el rendimiento.
* `key.converter` - establézcalo según los tipos de sus claves.
* `value.converter` - establézcalo según el tipo de datos de su topic. Estos datos no necesitan un esquema. El formato aquí debe ser coherente con el FORMAT especificado en el parámetro `http.api.url`. La opción más sencilla es usar JSON y el converter org.apache.kafka.connect.json.JsonConverter. También es posible tratar el valor como una cadena mediante el converter org.apache.kafka.connect.storage.StringConverter, aunque esto requerirá que el usuario extraiga un valor en la sentencia insert usando funciones. El [formato Avro](/docs/es/reference/formats/Avro/Avro) también es compatible con ClickHouse si se usa el converter io.confluent.connect.avro.AvroConverter.
Puede encontrar [aquí](https://docs.confluent.io/kafka-connect-http/current/connector_config.html) una lista completa de la configuración, incluida la forma de configurar un proxy, los reintentos y SSL avanzado.
Puede encontrar [aquí](https://github.com/ClickHouse/clickhouse-docs/tree/main/docs/integrations/data-ingestion/kafka/code/connectors/http_sink) archivos de configuración de ejemplo para los datos de muestra de GitHub, suponiendo que Connect se ejecuta en modo standalone y que Kafka está alojado en Confluent Cloud.
Asegúrese de que la tabla se ha creado. A continuación se muestra un ejemplo de un conjunto de datos mínimo de GitHub que utiliza un MergeTree estándar.
```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)
```
Inserte mensajes en Kafka. A continuación, usamos [kcat](https://github.com/edenhill/kcat) para insertar 10k mensajes.
```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
```
Una simple consulta a la tabla de destino "Github" debería confirmar la inserción de datos.
```sql theme={null}
SELECT count() FROM default.github;
| count\(\) |
| :--- |
| 10000 |
```
