Logstash - Input Kafka

 

Data를 손실없이 처리하는 Massege Queue방식의 카프카로 부터 안정적으로 Data를 받아 처리 할 수 있도록

kafka로부터 Data를 받는 방법에 대하여 알아보자.

 

input {
    kafka {
        bootstrap_servers => ["kafka:9092"] # kafka server & Port 지정
        topics => ["<<Your Topic Name>>"] # Topic 이름 지정
        codec => json #Data Codec 지정
        auto_offset_reset => "earliest" #Offset 지정 
        consumer_threads => 1 # Consumer 처리 Thread갯수 지정
        type => "kafka" # Plugin type지정 
        }
}

이중 기능이 좀 필요한 것들 몇개만 더 파보자

 

1. Auto_offset_reset 

• earliest: automatically reset the offset to the earliest offset
• latest: automatically reset the offset to the latest offset
• none: throw exception to the consumer if no previous offset is found for the consumer’s group
• anything else: throw exception to the consumer

2. Consumer_threads :

 Ideally you should have as many threads as the number of partitions for a perfect balance — more threads than partitions means that some threads will be idle

-> Thread 수에 따른 데이터 순서는 보장하지 않는다.

 

추가적인 옵션들이 다수 존재하나 필자는 위와 같은 방법으로 Data를 처리하였다. 

 

옵션에 대하여 좀더 보고싶다면 아래 링크에서 확인이 가능하다.

 

참조 : https://www.elastic.co/guide/en/logstash/current/plugins-inputs-kafka.html#plugins-inputs-kafka-consumer_threads