Using Vector with Kafka and ClickHouse

```
[sources.github]
type = "kafka"
auto_offset_reset = "smallest"
bootstrap_servers = "<kafka_host>:<kafka_port>"
group_id = "vector"
topics = [ "github" ]
tls.enabled = true
sasl.enabled = true
sasl.mechanism = "PLAIN"
sasl.username = "<username>"
sasl.password = "<password>"
decoding.codec = "json"

[sinks.clickhouse]
type = "clickhouse"
inputs = ["github"]
endpoint = "http://localhost:8123"
database = "default"
table = "github"
skip_unknown_fields = true
auth.strategy = "basic"
auth.user = "username"
auth.password = "password"
buffer.max_events = 10000
batch.timeout_secs = 1
```

A few important notes on this configuration and behavior of Vector:

- This example has been tested against Confluent Cloud. Therefore, the `sasl.*` and `ssl.enabled` security options may not be appropriate in self-managed cases.
- A protocol prefix isn't required for the configuration parameter `bootstrap_servers` e.g. `pkc-2396y.us-east-1.aws.confluent.cloud:9092`
- The source parameter `decoding.codec = "json"` ensures the message is passed to the ClickHouse sink as a single JSON object. If handling messages as Strings and using the default `bytes` value, the contents of the message will be appended to a field `message`. In most cases this will require processing in ClickHouse as described in the [Vector getting started](/docs/integrations/vector#4-parse-the-logs) guide.
- Vector [adds a number of fields](https://vector.dev/docs/reference/configuration/sources/kafka/#output-data) to the messages. In our example, we ignore these fields in the ClickHouse sink via the configuration parameter `skip_unknown_fields = true`. This ignores fields that aren't part of the target table schema. Feel free to adjust your schema to ensure these meta fields such as `offset` are added.
- Notice how the sink references of the source of events via the parameter `inputs`.
- Note the behavior of the ClickHouse sink as described [here](https://vector.dev/docs/reference/configuration/sinks/clickhouse/#buffers-and-batches). For optimal throughput, you may wish to tune the `buffer.max_events`, `batch.timeout_secs` and `batch.max_bytes` parameters. Per ClickHouse [recommendations](/docs/sql-reference/statements/insert-into#performance-considerations) a value of 1000 is should be considered a minimum for the number of events in any single batch. For uniform high throughput use cases, you may increase the parameter `buffer.max_events`. More variable throughputs may require changes in the parameter `batch.timeout_secs`
- The parameter `auto_offset_reset = "smallest"` forces the Kafka source to start from the start of the topic - thus ensuring we consume the messages published in step (1). You may require different behavior. See [here](https://vector.dev/docs/reference/configuration/sources/kafka/#auto_offset_reset) for further details.

4. Start Vector

```
vector --config ./kafka.toml
```

By default, a [health check](https://vector.dev/docs/reference/configuration/sinks/clickhouse/#healthcheck) is required before insertions begin to ClickHouse. This ensures connectivity can be established and the schema read. Prepend `VECTOR_LOG=debug`to obtain further logging which can be helpful should you encounter issues.

5. Confirm the insertion of the data.

```
SELECT count() AS count FROM github;
```

|  |  |
| --- | --- |
| count|  | | --- | | 200000 | |

- [Using Vector with Kafka and ClickHouse](#using-vector-with-kafka-and-clickhouse)
  - [License](#license)- [Gather your connection details](#gather-your-connection-details)- [Steps](#steps)

Was this page helpful?