[elasticsearch] 로그스태시로 엘라스틱서치 MySQL 동기화하기

[elasticsearch] 로그스태시로 엘라스틱서치 mysql 동기화하기

 

 

ELK Stack은 Elastic Search, Logstash, Kibana를 통칭하는 말인데, 데이터 로그들이 logstash를 거쳐 엘라스틱서치에 저장되고, 이를 시각화하여 Kibana에서 보여주게 된다.  그중에서도 로그스태시는 다양한 소스에서 데이터를 수집해 이를 변환하여, 저장소로 전달하는 역할을 한다. 소스의 종류, 사이즈, 형태에 관계 없이 수집할 수 있는 것이 특징적이다. 데이터가 로그 스태시를 거칠 때 필터링이 발생하는데, 필터를 통해 각 이벤트를 공통 형식으로 변환, 통합해줄 수 있다.

 

먼저, 로그스태시를 활용하여 jdbc(mysql)에 저장된 데이터를 엘라스틱서치로 자동으로 동기화하도록 설정하기를 해보자!

 

 

1️⃣  먼저 엘라스틱서치와 같은 버전의 Logstash를 다운받는다. (👉 다운로드 링크) 

 

2️⃣  jdbc 플러그인을 설치한다.

./bin/logstash-plugin install logstash-input-jdbc

3️⃣  jdbc 플러그인이 설치되면 자신이 사용하는 RDBMS의 connector jar 라이브러리를 설치한다.

mysql 의 경우 mysql-connector-java.jar를 다운 받아서 logstash의 lib 폴더에 넣어준다. (jar파일 다운받기)

 

4️⃣  다운받은 로그스태시에서 원하는 경로에 config 파일을 작성한다.(config안의 logstash-sample.conf 파일 참고)

input {
  jdbc {
    jdbc_driver_library => "C:/logstash-7.10.0/lib/mysql-connector-java-8.0.20.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://{호스트경로}/{db이름}"
    jdbc_user => "username"
    jdbc_password =>"password"
    schedule => "* * * * *"
    statement => "SELECT id, name, age FROM STUDENT"
  }
}
filter {
  mutate {
    remove_field => ["id", "@version"]
  }
}

output {
  elasticsearch {
      hosts => ["localhost:9200"]
      index => "index-name"
      document_id =>"%{id}"
  } stdout { 
	  codec => rubydebug 
  }
}

[input]  : MySql로 부터 가져올 부분에 대한 설정

- jdbc_driver_library : mysql-connector-java.jar를 다운받아 저장한 경로

- schedule : 스케쥴링 시간 (cron 표현식)

- statement : 실행할 쿼리문

[filter]: 데이터를 넣으면서 처리해줄 필터에 대한 설정

- remove_field : elasticsearch에 들어가지 않길 바라는 필드 처리

[output] : output을 elasticsearch로 보내는 설정

- hosts : elasticsearch 호스트 주소 설정

- index : elasticsearch 인덱스명

-document_id : elasticsearch로 보낼 document id를 지정

 

 

5️⃣ 로그스태시 실행

$sudo /logstash-7.10.0/bin/logstash -f /config/db.conf

 

 

이렇게 해주면, 스케쥴러가 돌아가며 원하는 시간마다 반복하여 mysql에 저장된 데이터를 엘라스틱서치로 넣어주게 된다.

로그스태시 최고 👍

 

 

 

 

 

참고  www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html