[Logstash] Elasticsearch 와 RDBMS 연동

Kibana > Stack Management > Logstash Pipelines

input {
	jdbc {
		jdbc_driver_library => "/usr/share/java/mysql-connector-java.jar"
		jdbc_driver_class => "com.mysql.jdbc.Driver"
		jdbc_connection_string => "jdbc:mysql://x.x.x.x:3011/Travel"
		jdbc_user => "id"
		jdbc_password => "pwd"
		statement_filepath => "/home/hunetdb/logstash/config/conf_sql/searchengine_travel.sql"
		schedule => "30 03,12 * * *"
		type => "search-engine-travel"
	}
}

filter {
	if [type] == "search-engine-travel" {
		mutate {
			remove_field => "message"
		}
    ruby {
			init => "require 'time'"
			code => "event.set('indexDate', Time.now.utc.getlocal.strftime('%Y.%m.%d'))"
		}
	}
}

output {
	if [type] == "search-engine-travel" {
		elasticsearch {
			hosts => ["x.x.x.10:9200", "x.x.x.11:9200", "x.x.x.12:9200"]
			index => "search-engine-travel-%{indexDate}"
			document_id => "%{travel_seq}"
			user => "elastic"
			password => "${elastic}"
			ssl => true
			ssl_certificate_verification => false
		}
	}
}

• input : 어떤 데이터를 수집할지 설정하는 부분
  • jdbc_driver_library : 사용하는 jdbc driver 위치 설정
  • jdbc_driver_class : 사용하는 jdbc driver 명시
  • jdbc_connection_string : 사용하는 db 종류, db가 위치한 host, db 이름 설정 ex) jdbc:mysql://x.x.x.x:3011/Travel
    • db 종류 : mysql
    • db host : x.x.x.x:3011
    • db 이름 : Travel
  • jdbc_user & jdbc_password : db 서버에서 권한을 부여 받은 id와 password
  • statement_filepath : db에서 어떤 query 및 프로시저를 실행할지 작성한 파일의 경로
• filter : 수집한 데이터를 가공
• output : 수집한 데이터를 전송
  • hosts : 데이터를 저장할 Elasticsearch의 Host
  • index : Elasticsearch Host의 어떤 Index에 저장할지 설정 (없으면 생성)
  • document_id : Document ID를 직접 지정

    • 미지정시 Elasticsearch가 임의값으로 생성

    • Document ID가 같으면 Update, 없으면 Insert

      Logstash와 JDBC를 사용해 RDBMS와 Elasticsearch의 동기화를 유지하는 방법