metricbeat对接kafka

article/2025/10/8 0:51:10

在监控系统中经常用到kafka来处理数据，上层平台会从kafka直接取数据进行分析，今天分享下metricbeat监控的数据如何输出到kafka中，并对各配置项进行详细说明。
metricbeat环境搭建点这里
kafka环境搭建点这里
-在metricbeat配置文件中配置output为kafka

output.kafka:# initial brokers for reading cluster metadatahosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]# message topic selection + partitioningtopic: '%{[type]}'partition.round_robin:reachable_only: falserequired_acks: 1compression: gzipmax_message_bytes: 1000000

配置说明：
Username：连接kafka的用户名（如果kafka有设置）
Password：连接kafka的密码
Topic：kafka中的topic设置
Key：发送消息时可以指定 key，producer根据此key和partition策略来选择存入哪个partition（可用于hash计算消息发布到哪个分区）
Partition：kafka的分发策略，有random（随机），round_robin（轮询），hash（哈希），默认是hash
Random.group_events:分发消息时对每个partition发送event的数量，默认为1（partition主要是实现负载均衡，让消息均匀的发送到多个partition中，设置为1则表示对一个partition发送一个event后开始对下一个partition发送event）
Round_robin.group_events:同上
Hash.hash：fields的列表，用于计算partition的hash值
Hash.random如果没有hash则采用随机分发的方式到partition
Client_id：theconfigurable ClientID used for logging, debugging, and auditing purposes. Thedefault is “beats”.
Worker：Thenumber of concurrent load-balanced Kafka output workers.
Metadata：kafka元数据更新设置（Topic/Partion与broker的映射关系：每一个Topic的每一个Partion，得知道其对应的broker列表是什么，其中leader是谁，follower是谁）
Refresh_frequency:刷新频率，默认10分钟
Retry.max：尝试更新次数，默认3
Retry.backoff:每次尝试更新之间的间隔，默认250ms
Max_retries:消息发布失败后重新尝试的次数，默认3（beat->kafka）
bulk_max_size:单个kafka请求里面的最大事件数，默认2048
The maximum number of events to bulk in a singleKafka request. The default is 2048
timeout：等待kafka broker响应的时间，默认30s
broker_timeout：kafka broker等待请求的最大时长，默认10s
channel_buffer_size:每个kafka broker在输出管道中的消息缓存数，默认256
Per Kafka broker number of messages buffered inoutput pipeline. The default is 256.
keep_alive：网络连接的保活时间，默认为0，不开启保活机制
compression：输出压缩码，可选项有none，snappy，gzip，默认为gzip
（kafka支持的压缩，数据会先被压缩，然后被生产者发送，并且在服务端也是保持压缩状态，只有在最终的消费者端才会被解压缩）。
max_message_bytes：允许的最大json消息大小，默认为1000000，超出的会被丢弃，应该小于broker的 message.max.bytes（broker能接收消息的最大字节数）
required_acks:kafka的响应返回值，0位无等待响应返回，继续发送下一条消息；1表示等待本地提交（leader broker已经成功写入，但follower未写入），-1表示等待所有副本的提交，默认为1
flush_interval:在producer api调用时，等待新事件的时间秒数
The number of seconds to wait for new eventsbetween two producer API calls.
ssl：ssl的配置项，如对kafka连接的root CA等，具体配置，参考点这里

-选择配置：

vi /etc/metricbeat/metricbeat.yml
output.kafka:#Array of hosts to connect to.hosts: ["192.168.1.100:9092"]#message topictopic: "test2topic"

然后重启metricbeat服务即可。
在kafka中查看topic中保存的消息：

{"@timestamp":"2017-02-20T02:48:31.499Z","beat":{"hostname":"centos-78","name":"centos-78","version":"5.2.0"},"metricset":{"module":"system","name":"process","rtt":72121},"system":{"process":{"cmdline":"pickup-l -t unix -u","cpu":{"start_time":"2017-02-20T02:42:40.000Z","total":{"pct":0.000000}},"fd":{"limit":{"hard":4096,"soft":4096},"open":12},"memory":{"rss":{"bytes":4059136,"pct":0.001400},"share":3035136,"size":91222016},"name":"pickup","pgid":2139,"pid":4847,"ppid":2139,"state":"sleeping","username":"postfix