如何提升Logstash并行度且不破坏Kafka分区数据顺序？

场景说明 #

使用Logstash实现Kafka到Elasticsearch的同步，目标Kafka主题包含8个分区，已配置consumer_threads=8实现并行消费，输入段配置如下：

input {
  kafka {
    bootstrap_servers => "bootstrapServer"
    topics => "topicName"
    codec => json
    group_id => "groupName"
    id => ""
    consumer_threads => 8
  }
}

后续包含filter与Output处理环节，针对该场景的技术问题解答如下：

问题1：如何提升Logstash worker并行度，且不影响Kafka分区内的数据顺序？ #

• 调整pipeline.workers参数：该参数控制filter和Output阶段的并行线程数，直接调大会打乱单分区数据顺序，需配合分区感知的有序队列路由。
• 配置有序队列：
  • 若使用默认内存队列，在logstash.yml中设置pipeline.ordered: true；
  • 若使用持久化队列（queue.type: persisted），设置partition.ordered: true。
• 注意事项：保持consumer_threads与Kafka分区数匹配（当前已设为8），pipeline.workers可设为等于或大于分区数，通过有序队列路由，确保同一Kafka分区的事件始终由同一个worker线程处理，既提升并行度，又不破坏分区内数据顺序。

问题2：Logstash在输入与filter、Output之间是否使用内存队列？如何确保单个Kafka分区的数据由Logstash的单个filter和Output线程处理？ #

• 队列机制：默认使用内存队列（queue.type: memory），也可配置为持久化队列（queue.type: persisted），避免进程崩溃导致事件丢失。
• 保证单分区事件由单个线程处理的方法：
  1. 开启有序路由：内存队列设置pipeline.ordered: true，持久化队列设置partition.ordered: true（均在logstash.yml中配置）；
  2. 匹配消费线程数：保持consumer_threads等于Kafka分区数，每个消费线程对应一个分区，队列会将同一分区的事件路由至同一个worker线程；
  3. 避免无序插件：不要使用会打乱事件顺序的过滤器插件，确保分区内事件在处理链路中保持串行。

内容的提问来源于stack exchange，提问作者Jobs