Conector HTTP Sink da Confluent

• Autogerenciado: Baixe o pacote da Confluent e instale-o localmente. Siga as instruções de instalação do conector, conforme documentado aqui. Se você usar o método de instalação confluent-hub, seus arquivos de configuração locais serão atualizados.
• Confluent Cloud: Uma versão totalmente gerenciada do HTTP Sink está disponível para quem usa o Confluent Cloud para hospedar o Kafka. Isso exige que seu ambiente ClickHouse esteja acessível a partir da Confluent Cloud.

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

• Informe o nome do tópico que você criou
• Autenticação
  • HTTP Url - URL do ClickHouse Cloud com uma consulta INSERT especificada: <protocol>://<clickhouse_host>:<clickhouse_port>?query=INSERT%20INTO%20<database>.<table>%20FORMAT%20JSONEachRow. Observação: a consulta deve ser codificada.
  • Endpoint Authentication type - BASIC
  • Auth username - nome de usuário do ClickHouse
  • Auth password - senha do ClickHouse

• Configuração
  • Input Kafka record value format - Depende dos seus dados de origem, mas, na maioria dos casos, será JSON ou Avro. Assumimos JSON nas configurações a seguir.
  • Na seção advanced configurations:
    • HTTP Request Method - Defina como POST
    • Request Body Format - json
    • Batch batch size - De acordo com as recomendações do ClickHouse, defina esse valor como no mínimo 1000.
    • Batch json as array - true
    • Retry on HTTP codes - 400-500, mas ajuste conforme necessário; por exemplo, isso pode mudar se você tiver um proxy HTTP na frente do ClickHouse.
    • Maximum Reties - o padrão (10) é adequado, mas fique à vontade para ajustar se quiser tentativas de repetição mais robustas.

O conector HTTP Sink não agrupa em lote solicitações de mensagens que contêm valores de header do Kafka diferentes.

1. Verifique se os registros do Kafka têm a mesma chave.
2. Ao adicionar parâmetros à URL da API HTTP, cada registro pode resultar em uma URL exclusiva. Por esse motivo, o agrupamento em lote é desativado ao usar parâmetros de URL adicionais.

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)

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

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

• request.method - Defina como POST
• retry.on.status.codes - Defina como 400-500 para repetir a tentativa em qualquer código de erro. Ajuste conforme os erros esperados nos dados.
• request.body.format - Na maioria dos casos, será JSON.
• auth.type - Defina como BASIC se você usar segurança com o ClickHouse. Outros mecanismos de autenticação compatíveis com o ClickHouse não são compatíveis no momento.
• ssl.enabled - defina como true se estiver usando SSL.
• connection.user - nome de usuário do ClickHouse.
• connection.password - senha do ClickHouse.
• batch.max.size - O número de linhas a serem enviadas em um único lote. Certifique-se de definir um número adequadamente alto. De acordo com as recomendações do ClickHouse, um valor de 1000 deve ser considerado o mínimo.
• tasks.max - O conector HTTP Sink oferece suporte à execução de uma ou mais tarefas. Isso pode ser usado para aumentar o desempenho. Junto com o tamanho do lote, esse é o principal meio de melhorar o desempenho.
• key.converter - defina de acordo com os tipos das suas chaves.
• value.converter - defina com base no tipo de dados no seu tópico. Esses dados não precisam de um esquema. O formato aqui deve ser consistente com o FORMAT especificado no parâmetro http.api.url. A opção mais simples é usar JSON e o conversor org.apache.kafka.connect.json.JsonConverter. Também é possível tratar o valor como uma string, por meio do conversor org.apache.kafka.connect.storage.StringConverter, embora isso exija que o usuário extraia um valor na instrução insert usando funções. O formato Avro também é compatível com o ClickHouse ao usar o conversor io.confluent.connect.avro.AvroConverter.

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)

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

SELECT count() FROM default.github;

| count\(\) |
| :--- |
| 10000 |