腾讯云Logstash实战4-使用Logstash消费kafka
2021-12-25 本文已影响0人
bellengao
Logstash的一个典型应用场景,就是消费kafka中的数据并且写入到Elasticsearch, 使用腾讯云Logstash产品,可以通过简单的配置快速地完成这一过程。
创建管道
在“管道管理”页面,点击“新建管道”按钮,创建一个管道:
image
进入管道配置页面,点击“引用模板”按钮,同时引用“input-kafka”和“output-elasticsearch”两个模板:
image
image
在管道配置中,分别针对“input-kafka”和“output-elasticsearch”进行配置,一些关键的配置参数说明如下:
input-kafka
- bootstrap_servers:kafka服务端地址列表
- client_id: 客户端ID
- group_id: 消费组ID
- consumer_threads: 消费线程数量,最好保持该参数*Logstash实例节点的数量=topic的partitions数量
- topics: topic列表
- auto_offset_reset: 当kafka中topic没有初始的offset时,如何重置offset,常用可选值为earliest(最早)、latest(最新)
- type: 标识字段
查看更多参数,可以参考input-kafka
output-elasticsearch
- hosts: elasticsearch集群地址列表
- user: elasticsearch集群账号
- password: elasticsearch集群密码
- index: 索引名称
- document_type: 索引type,对于不同版本的ES集群,该字段有不同的默认值,5.x及以下的集群,默认会使用input中指定的type字段,如果type字段不存在,则该字段的值为doc;6.x的集群,该字段默认值为doc;7.x的集群,该字段默认值为_doc; 8.x的集群,不会使用该字段
- document_id: 文档ID
查看更多参数,可以参考output-elasticsearch
在配置完管道后,点击“保存并部署”创建一个管道并自动部署:
image
查看日志
在控制台查看Logstash的运行日志,如果没有ERROR级别的日志,则说明管道运行正常:
image
查看数据写入情况
进入到output-elasticsearch中定义的输出端的ES集群对应的kibana页面,在Dev tools工具栏里查看索引是否存在,以及索引的文档数量是否正确:
image