Logstash(Shipper) 를 사용해 Kafka 메세지를 elasticsearch로 저장

1. download

Logstash를 다운받아 압축을 푼다.

https://www.elastic.co/kr/downloads/logstash

현재 최신 버전인 logstash-5.1.2 버전을 다운받는다. 보통 elasticsearch도 같은 버전으로 설치를 한다.

2. configuration

logstash를 실행할때 지정한 config 파일에 따라 하나의 프로세스로 동작을 한다.

우리가 할껀 input을 kafka로 받아서 output을 elasticsearch로 보내는 것이 대한 config파일을 작성한다.

logstash-metric.conf

input { kafka { bootstrap_servers => "192.168.0.16:9092,192.168.0.18:9092" topics => "metric" group_id => "metric-group-logstash" consumer_threads => 2 } } filter { json { source => "message" } mutate { remove_field => ["message"] } } output { elasticsearch { hosts => ["192.168.0.18:9200"] index => "metricbeat-%{+YYYY.MM.dd}" document_type => "metric" } stdout { codec => rubydebug } }

이 파일을 어디든 둬도 상관 없는데 이 예제에서는 logstash하위 config폴더에 저장한다. 

3. kafka 버전에 따라 dependency 변경

logstash-5.1.2는 kafka 10.x 를 기본으로 사용하도록 설정이 되어있다. 이 예제에서는 kafka 0.9.x 버전을 사용하기 때문에 약간의 dependency 설정을 바꿔줘야한다.

설치된 logstash 폴더에서 아래 폴더를 보면 kafka-client-10.x.jar 가 있을 것이다. 이것을 사용하고 있는 kafka 버전에 맞는 것으로 바꿔줘야 한다.

/logstash-5.1.2/vendor/bundle/jruby/1.9/gems/logstash-input-kafka-5.1.2/vendor/jar-dependencies/runtime-jars/

/logstash-5.1.2/vendor/bundle/jruby/1.9/gems/logstash-output-kafka-5.1.2/vendor/jar-dependencies/runtime-jars/

0.9.0.1 의 kafka-clients는 아래 url에서 받으면 된다.

http://central.maven.org/maven2/org/apache/kafka/kafka-clients/0.9.0.1/kafka-clients-0.9.0.1.jar

간단히 input, output kafka dependency 폴더에서 10.x버전을 지우고 0.9.x 를 복사하는 것이다.

4. 실행

kafka에 있는 message는 metricbeat로 저장된 pc의 리소스 데이터이다.

> sh bin/logstash -f config/logstash-metric.conf

위 명령어를 실행하면 아래 그림처럼 실행 설정이 나타나고 Logstash가 실행된다. output에 stdout를 설정했기 때문에 콘솔에서도 메세지가 표시된다.

5. 데이터 확인 

Kibana를 통해 메세지가 잘 들어가는지 확인해 본다

끝