Input 配置示例
以下配置可以实现对 kafka 读取端(consumer)的基本使用。
消费端更多详细的配置请查看 http://kafka.apache.org/documentation.html#consumerconfigs kafka 官方文档的消费者部分配置文档。
input {
kafka {
zk_connect => "localhost:2181"
group_id => "logstash"
topic_id => "test"
codec => plain
reset_beginning => false # boolean (optional), default: false
consumer_threads => 5 # number (optional), default: 1
decorate_events => true # boolean (optional), default: false
}
}
Input 解释
作为 Consumer 端,插件使用的是 High-level-consumer API
,请结合上述 kafka 基本概念进行设置:
- group_id
消费者分组,可以通过组 ID 去指定,不同的组之间消费是相互不受影响的,相互隔离。
- topic_id
指定消费话题,也是必填项目,指定消费某个 topic
,这个其实就是订阅某个主题,然后去消费。
- reset_beginning
logstash 启动后从什么位置开始读取数据,默认是结束位置,也就是说 logstash 进程会以从上次读取结束时的偏移量开始继续读取,如果之前没有消费过,那么就开始从头读取.如果你是要导入原有数据,把这个设定改成 "true", logstash 进程就从头开始读取.有点类似 cat
,但是读到最后一行不会终止,而是变成tail -F
,继续监听相应数据。
- decorate_events
在输出消息的时候会输出自身的信息包括:消费消息的大小, topic 来源以及 consumer 的 group 信息。
- rebalance_max_retries
当有新的 consumer(logstash) 加入到同一 group 时,将会 reblance
,此后将会有 partitions
的消费端迁移到新的 consumer
上,如果一个 consumer
获得了某个 partition
的消费权限,那么它将会向zookeeper
注册, Partition Owner registry
节点信息,但是有可能此时旧的 consumer
尚没有释放此节点,此值用于控制,注册节点的重试次数。
- consumer_timeout_ms
指定时间内没有消息到达就抛出异常,一般不需要改。
以上是相对重要参数的使用示例,更多参数可以选项可以跟据 https://github.com/joekiller/logstash-kafka/blob/master/README.md 查看 input 默认参数。
注意
1.想要使用多个 logstash 端协同消费同一个 topic
的话,那么需要把两个或是多个 logstash 消费端配置成相同的 group_id
和 topic_id
, 但是前提是要把相应的 topic 分多个 partitions (区),多个消费者消费是无法保证消息的消费顺序性的。
这里解释下,为什么要分多个 partitions(区), kafka 的消息模型是对 topic 分区以达到分布式效果。每个
topic
下的不同的 partitions (区)只能有一个 Owner 去消费。所以只有多个分区后才能启动多个消费者,对应不同的区去消费。其中协调消费部分是由 server 端协调而成。不必使用者考虑太多。只是消息的消费则是无序的。
总结:保证消息的顺序,那就用一个 partition。 kafka 的每个 partition 只能同时被同一个 group 中的一个 consumer 消费。
Output 配置
以下配置可以实现对 kafka 写入端 (producer) 的基本使用。
生产端更多详细的配置请查看 http://kafka.apache.org/documentation.html#producerconfigs kafka 官方文档的生产者部分配置文档。
output {
kafka {
broker_list => "localhost:9092"
topic_id => "test"
compression_codec => "snappy" # string (optional), one of ["none", "gzip", "snappy"], default: "none"
}
}
Output 解释
作为 Producer 端使用,以下仅为重要概念解释,请结合上述 kafka 基本概念进行设置:
- compression_codec
消息的压缩模式,默认是 none,可以有 gzip 和 snappy (暂时还未测试开启压缩与不开启的性能,数据传输大小等对比)。
- compressed_topics
可以针对特定的 topic 进行压缩,设置这个参数为 topic
,表示此 topic
进行压缩。
- request_required_acks
消息的确认模式:
可以设置为 0: 生产者不等待 broker 的回应,只管发送.会有最低能的延迟和最差的保证性(在服务器失败后会导致信息丢失) 可以设置为 1: 生产者会收到 leader 的回应在 leader 写入之后.(在当前 leader 服务器为复制前失败可能会导致信息丢失) 可以设置为 -1: 生产者会收到 leader 的回应在全部拷贝完成之后。
- partitioner_class
分区的策略,默认是 hash 取模
- send_buffer_bytes
socket 的缓存大小设置,其实就是缓冲区的大小
消息模式相关
- serializer_class
消息体的系列化处理类,转化为字节流进行传输,请注意 encoder 必须和下面的 key_serializer_class
使用相同的类型。
- key_serializer_class
默认的是与 serializer_class
相同
- producer_type
生产者的类型 async
异步执行消息的发送 sync
同步执行消息的发送
- queue_buffering_max_ms
异步模式下,那么就会在设置的时间缓存消息,并一次性发送
- queue_buffering_max_messages
异步的模式下,最长等待的消息数
- queue_enqueue_timeout_ms
异步模式下,进入队列的等待时间,若是设置为0,那么要么进入队列,要么直接抛弃
- batch_num_messages
异步模式下,每次发送的最大消息数,前提是触发了 queue_buffering_max_messages
或是queue_enqueue_timeout_ms
的限制
以上是相对重要参数的使用示例,更多参数可以选项可以跟据 https://github.com/joekiller/logstash-kafka/blob/master/README.md 查看 output 默认参数。
小贴士
logstash-kafka 插件输入和输出默认 codec
为 json 格式。在输入和输出的时候注意下编码格式。消息传递过程中 logstash 默认会为消息编码内加入相应的时间戳和 hostname 等信息。如果不想要以上信息(一般做消息转发的情况下),可以使用以下配置,例如:
output {
kafka {
codec => plain {
format => "%{message}"
}
}
}
作为 Consumer 从kafka中读数据,如果为非 json 格式的话需要进行相关解码,例如:
input {
kafka {
zk_connect => "xxx:xxx"
group_id => "test"
topic_id => "test-topic"
codec => "line"
...........
}
}
性能
队列监控
其实 logstash 的 kafka 插件性能并不是很突出,可以通过使用以下命令查看队列积压消费情况:
$/bin/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group test
队列积压严重,性能跟不上的情况下,结合实际服务器资源,可以适当增加 topic 的 partition 多实例化 Consumer 进行消费处理消息。
input-kafka 的 JSON 序列化性能
此外,跟 logstash-input-syslog 改在 filter 阶段 grok 的优化手段类似,也可以将 logstash-input-kafka 的默认 JSON 序列化操作从 codec 阶段后移到 filter 阶段。如下:
input {
kafka {
codec => plain
}
}
filter {
json {
source => "message"
}
}
然后通过 bin/logstash -w $num_cpus
运行,利用多核计算优势,可以获得大约一倍左右的性能提升。