跳到主要内容
跳到主要内容

kafka-vector

使用 Vector 与 Kafka 和 ClickHouse​

Vector 是一个与供应商无关的数据管道，能够从 Kafka 读取数据并将事件发送到 ClickHouse。

一个关于 Vector 与 ClickHouse 的 入门指南 重点介绍了日志用例和从文件读取事件。我们利用 Github 示例数据集，该数据集包含位于 Kafka 主题上的事件。

Vector 使用 sources 通过推送或拉取模型来检索数据。Sinks 则提供事件的目标。因此，我们使用 Kafka 源和 ClickHouse 接收器。请注意，虽然 Kafka 被支持为接收器，但 ClickHouse 源目前不可用。因此，Vector 不适合希望将数据从 ClickHouse 传输到 Kafka 的用户。

Vector 还支持数据的 transformation。这超出了本指南的范围。如果用户需要这方面的帮助，请参考 Vector 文档。

请注意，当前 ClickHouse 接收器的实现利用了 HTTP 接口。目前 ClickHouse 接收器不支持使用 JSON schema。数据必须以纯 JSON 格式或字符串格式发布到 Kafka。

许可证​

Vector 根据 MPL-2.0 许可证 分发。

收集您的连接详情​

要通过 HTTP(S) 连接到 ClickHouse，您需要以下信息：

  • HOST 和 PORT：通常，当使用 TLS 时，端口为 8443；当不使用 TLS 时，端口为 8123。

  • 数据库名称：开箱即用时，有一个名为 default 的数据库，请使用您要连接的数据库名称。

  • 用户名和密码：开箱即用时，用户名为 default。请使用适合您用例的用户名。

您的 ClickHouse Cloud 服务详细信息可在 ClickHouse Cloud 控制台中找到。 选择您要连接的服务并点击 Connect：

ClickHouse Cloud 服务连接按钮

选择 HTTPS，详细信息会在示例 curl 命令中提供。

ClickHouse Cloud HTTPS 连接详细信息

如果您使用的是自管理的 ClickHouse，连接详细信息由您的 ClickHouse 管理员设置。

步骤​

  1. 创建 Kafka github 主题，并插入 Github 数据集。

该数据集包含 200,000 行，专注于 ClickHouse/ClickHouse 存储库。

  1. 确保目标表已创建。下面我们使用默认数据库。
  1. 下载并安装 Vector。创建一个 kafka.toml 配置文件，并修改 Kafka 和 ClickHouse 实例的值。

关于此配置和 Vector 行为的一些重要注意事项：

  • 此示例已在 Confluent Cloud 上进行了测试。因此，sasl.* 和 ssl.enabled 安全选项在自管理的情况下可能不合适。
  • 配置参数 bootstrap_servers 不需要协议前缀，例如 pkc-2396y.us-east-1.aws.confluent.cloud:9092
  • 源参数 decoding.codec = "json" 确保消息被作为单个 JSON 对象传递给 ClickHouse 接收器。如果将消息作为字符串处理并使用默认的 bytes 值，则消息内容将附加到字段 message。在大多数情况下，这将需要在 ClickHouse 中处理，如 Vector 入门指南 所描述。
  • Vector 为消息添加了许多字段。在我们的示例中，我们通过配置参数 skip_unknown_fields = true 忽略这些字段。这将忽略不属于目标表架构的字段。请根据需要调整您的架构，以确保添加诸如 offset 的元字段。
  • 请注意，接收器如何通过参数 inputs 引用事件源。
  • 请注意 ClickHouse 接收器的行为如 此处所述。为了获得最佳吞吐量，用户可能希望调整 buffer.max_events、batch.timeout_secs 和 batch.max_bytes 参数。根据 ClickHouse 建议，单个批次中的事件数量应考虑为 1000 的最小值。对于均匀的高吞吐量用例，用户可以增加 buffer.max_events 参数。更多变量的吞吐量可能需要调整 batch.timeout_secs 参数。
  • 参数 auto_offset_reset = "smallest" 强制 Kafka 源从主题的开始处启动，从而确保我们消费在步骤 (1) 发布的消息。用户可能需要不同的行为。有关详细信息，请参见 这里。
  1. 启动 Vector

默认情况下，在开始插入到 ClickHouse 之前需要进行 健康检查。这确保可以建立连接并读取架构。前面加上 VECTOR_LOG=debug 以获取进一步的日志，这在遇到问题时可能会有所帮助。

  1. 确认数据的插入。
count
200000