Kafka

Kafka

如果你使用 Kafka 或兼容 Kafka 的消息队列进行可观测性数据传输，可以直接将数据写入到 GreptimeDB 中。

这里我们使用 Vector 作为工具将数据从 Kafka 传输到 GreptimeDB。

指标

从 Kafka 写入指标到 GreptimeDB 时，消息应采用 InfluxDB 行协议格式。例如：

1. census,location=klamath,scientist=anderson bees=23 1566086400000000000

然后配置 Vector 使用 influxdb 解码器来处理这些消息。

1. [sources.metrics_mq]
2. # 指定源类型为 Kafka
3. type = "kafka"
4. # Kafka 的消费者组 ID
5. group_id = "vector0"
6. # 要消费消息的 Kafka 主题列表
7. topics = ["test_metric_topic"]
8. # 要连接的 Kafka 地址
9. bootstrap_servers = "kafka:9092"
10. # `influxdb` 表示消息应采用 InfluxDB 行协议格式
11. decoding.codec = "influxdb"

13. [sinks.metrics_in]
14. inputs = ["metrics_mq"]
15. # 指定接收器类型为 `greptimedb_metrics`
16. type = "greptimedb_metrics"
17. # GreptimeDB 服务器的端点
18. # 将 <host> 替换为实际的主机名或 IP 地址
19. endpoint = "<host>:4001"
20. dbname = "<dbname>"
21. username = "<username>"
22. password = "<password>"
23. tls = {}

有关 InfluxDB 行协议指标如何映射到 GreptimeDB 数据的详细信息，请参阅 InfluxDB 行协议文档中的数据模型部分。

日志

开发人员通常处理两种类型的日志：JSON 日志和纯文本日志。例如以下从 Kafka 发送的日志示例。

纯文本日志：

1. 127.0.0.1 - - [25/May/2024:20:16:37 +0000] "GET /index.html HTTP/1.1" 200 612 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"

或 JSON 日志：

1. {
2. "timestamp": "2024-12-23T10:00:00Z",
3. "level": "INFO",
4. "message": "Service started"
5. }

GreptimeDB 将这些日志转换为具有多个列的结构化数据，并自动创建必要的表。 Pipeline 在写入到 GreptimeDB 之前将日志处理为结构化数据。不同的日志格式需要不同的 Pipeline 来解析，详情请继续阅读下面的内容。

JSON 格式的日志

对于 JSON 格式的日志（例如 {"timestamp": "2024-12-23T10:00:00Z", "level": "INFO", "message": "Service started"}），你可以使用内置的 greptime_identity pipeline 直接写入日志。此 pipeline 根据 JSON 日志消息中的字段自动创建列。

你只需要配置 Vector 的 transforms 设置以解析 JSON 消息，并使用 greptime_identity pipeline，如以下示例所示：

1. [sources.logs_in]
2. type = "kafka"
3. # Kafka 的消费者组 ID
4. group_id = "vector0"
5. # 要消费消息的 Kafka 主题列表
6. topics = ["test_log_topic"]
7. # 要连接的 Kafka 代理地址
8. bootstrap_servers = "kafka:9092"

10. # 将日志转换为 JSON 格式
11. [transforms.logs_json]
12. type = "remap"
13. inputs = ["logs_in"]
14. source = '''
15. . = parse_json!(.message)
16. '''

18. [sinks.logs_out]
19. # 指定此接收器将接收来自 `logs_json` 源的数据
20. inputs = ["logs_json"]
21. # 指定接收器类型为 `greptimedb_logs`
22. type = "greptimedb_logs"
23. # GreptimeDB 服务器的端点
24. endpoint = "http://<host>:4000"
25. compression = "gzip"
26. # 将 <dbname>、<username> 和 <password> 替换为实际值
27. dbname = "<dbname>"
28. username = "<username>"
29. password = "<password>"
30. # GreptimeDB 中的表名，如果不存在，将自动创建
31. table = "demo_logs"
32. # 使用内置的 `greptime_identity` 管道
33. pipeline_name = "greptime_identity"

文本格式的日志

对于文本格式的日志，例如下面的访问日志格式，你需要创建自定义 pipeline 来解析它们：

1. 127.0.0.1 - - [25/May/2024:20:16:37 +0000] "GET /index.html HTTP/1.1" 200 612 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"

创建 pipeline

要创建自定义 pipeline，请参阅创建 pipeline 和 pipeline 配置文档获取详细说明。

写入数据

创建 pipeline 后，将其配置到 Vector 配置文件中的 pipeline_name 字段。

1. # sample.toml
2. [sources.log_mq]
3. # 指定源类型为 Kafka
4. type = "kafka"
5. # Kafka 的消费者组 ID
6. group_id = "vector0"
7. # 要消费消息的 Kafka 主题列表
8. topics = ["test_log_topic"]
9. # 要连接的 Kafka 地址
10. bootstrap_servers = "kafka:9092"

12. [sinks.sink_greptime_logs]
13. # 指定接收器类型为 `greptimedb_logs`
14. type = "greptimedb_logs"
15. # 指定此接收器将接收来自 `log_mq` 源的数据
16. inputs = [ "log_mq" ]
17. # 使用 `gzip` 压缩以节省带宽
18. compression = "gzip"
19. # GreptimeDB 服务器的端点
20. # 将 <host> 替换为实际的主机名或 IP 地址
21. endpoint = "http://<host>:4000"
22. dbname = "<dbname>"
23. username = "<username>"
24. password = "<password>"
25. # GreptimeDB 中的表名，如果不存在，将自动创建
26. table = "demo_logs"
27. # 你创建的自定义管道名称
28. pipeline_name = "your_custom_pipeline"

Demo

有关数据转换和写入的可运行演示，请参阅 Kafka Ingestion Demo。