MariaDB-Logstash-elasticsearch 연동하기

Logstash로 MariaDB를 접속하여 데이터를 수집하고 elasticsearch에 연동하는 방법입니다. 

1. Logstash conf 파일 작성

input {
  jdbc {
    jdbc_driver_library => "C:/dev/logstash-7.14.1/lib/mariadb-java-client-2.7.1.jar"
    jdbc_driver_class => "org.mariadb.jdbc.Driver"
    jdbc_connection_string => "jdbc:mariadb://ip:port/dbname?useUnicode=true&characterEncoding=utf-8"
    jdbc_user => "username"
    jdbc_password => "password"
    schedule => "* * * * *"
    statement => "SELECT 
                    BOOK_ID, TITLE, MIN_AGE, MAX_AGE, PUBLISHER, DEPARTMENT, STATE, SHIPPING_FEE, REG_DATE, NEW_OR_USED, LIST_PRICE, PRICE, PUB_YEAR, SELLER_NAME, SELLER_CONTACT 
                  FROM
                    combook
                  WHERE 
                     REG_DATE > DATE_FORMAT(:sql_last_value,'%Y%m%d%H%i%s')  
                  AND
                    DELETE_YN = 'n'
                  "
    use_column_value => true
    tracking_column => "reg_date"
    tracking_column_type => "timestamp"
    record_last_run => true
    clean_run => true
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "combook_%{+YYYY.MM.dd}"
  }
}

제가 작성한 combook.conf 입니다. 

• input 모듈은 데이터를 Logstash에 input해주는 역할을 합니다. 데이터베이스의 데이터를 수집하는 기능으로서 CRON 구문을 이용해서 주기적으로 스케줄링하여 쿼리를 실행하고, 결과 집합의 단일행 단위로 이벤트되어 처리된다. 열은 필드로 변환되어 처리된다. 드라이버를 별도로 지원하지 않고 외부라이브러리를 명시적으로 설정하여 사용할 수 있다. 스케줄은 rufus-scheduler의 구문을 따른다. 
• jdbc_driver_library: 데이터베이스 접속을 위해 데이터베이스 제작사에서 배포하는 라이브러리. jdbc jar 파일의 경로를 지정해준다.
• jdbc_driver_class: jdbc jar파일에 포함된 자바 클래스 중 메인클레스에 해당하는 Driver 클래스를 설정. 
• jdbc_connection_string: 데이터베이스에 접속하기 위해 필요한 접속정보. DB접속 url.
• jdbc_user: 데이터베이스에서 사용할 사용자 정보. DB접속 계정.
• jdbc_password: 데이터베이스에서 사용할 사용자 정보. DB접속 계정의 password.
• schedule: 데이터를 접속해서 조회할 수 있는 주기를 설정한다. 스케줄 방식은 rufus-scheduler 구문을 따른다. 
• statement: 데이터를 조회하는 SQL 문을 입력값으로 설정한다. sql_last_value를 사용하여 직전에 마지막으로 처리된 일련번호나 날짜를 대입한다. 
• use_column_value: 최종 데이터를 기록하기 위한 변수(sql_last_value)를 사용할지 여부를 입력. true/false
• tracking_column: use_column_value의 기준이 될 컬럼명 
• tracking_column_type: tracking_column의 type. timestamp나 numeric타입을 지원. 
• record_last_run: 마지막 처리 상태를 저장할지 여부를 입력.
• clean_run: true로 설정하면 숫자일경우"0", 날짜일경우 "1970.01.01"로 초기화
• output 모듈은 데이터의 조회, 정제, 가공등의 작업이 마무리된 후에 데이터의 저장을 담당하는 플러그인이다. 
• host: 연동하고자 할 elasticsearch의 url을 입력합니다. 
• index: 수집된 데이터가 사용될 인덱스명을 기입합니다. 저는 combook_오늘날짜가 오도록 설정했습니다. 

2. Logstash 실행

combook.conf 파일 작성이 완료되었으니, Logstash를 실행하여 DB의 데이터를 수집합니다. 

 

먼저, elasticsearch, kibana를 실행시키고 난 후,

관리자권한으로 명령 프롬프트를 실행시키고, cd 명령어로 Logstash설치경로 밑에 bin 으로 이동합니다. 

cd C:\dev\logstash-7.14.1\bin

다음으로, Logstash를 실행합니다. 

logstash.bat -f combook.conf

일련의 로그들이 나오고 하단에 statement에 입력된 SQL문에 변수가 대입되어서 실행중인것을 알 수 있습니다. 

정상적으로 수집이 완료되면 elasticsearch cmd에 찍히는 로그로 확인할 수 있습니다. 

create되고 update가 한번 되었네요, conf파일에 schedule 부분에 * * * 로 작성되어 있어서 1분마다 작업이 시행되어서 그렇습니다. 수집이 완료되었으니 Ctrl + c 를 눌러서 Logstash를 종료합니다. 

3. 수집 결과 확인

수집 결과 확인을 위해 Kibana에 접속합니다. 

url: http://localhost:5601

 

좌측 메뉴를 열어서 최 하단에 Stack Management를 선택합니다. 

수집이 정상적으로 이루어졌다면, 방금 수집한 데이터를 확인해 볼 수 있습니다. 

 

4. Index Pattern 만들기 

인덱스 패턴이란

엘라스틱서치는 인덱스로 데이터를 저장한다. 관계형 데이터베이스를 더 잘 알고 있다면 테이블과 비슷하다고 생각하면 된다. 인덱스 패턴은 사용자가 조사하기 원하는 엘라스틱서치 인덱스를 키바나에 알려준다. 

엘라스틱서치에서 특정 인덱스를 대상으로 인덱스 패턴을 생성할 수 있다. 또는 와일드카드 문자인 *를 사용해 동시에 여러 인덱스를 쿼리할 수 있다. (데이터베이스에 여러 테이블을 구현하듯) 키바나에 여러 인덱스 패턴을 구현할 수 있다. 시각화를 생성하거나 데이터를 검색할 경우, 검색을 실행할 인덱스 패턴을 선택해야 한다.

수집이 정상적으로 이루어졌다면, 방금 수집한 데이터를 확인해 볼 수 있습니다. 

수집이 잘 되었다면 위에 사진에 표시된 Index Patterns로 이동하고, 우측에 Create index pattern을 선택합니다. 

Index pattern name에 우리가 index에 지정해준 이름을 기입합니다. 끝에 *이 자동으로 붙기 때문에 combook_날짜 형식으로 인덱스명을 만들었기 때문에 combook_까지만 입력하였습니다. 

입력이 완료되면 Next step을 선택합니다. 

 

데이터에 시간 관련 필드가 있는 분은 Time field로 지정할 수 있습니다. 저는 @timestamp로 지정하였습니다. 선택이 완료되면 Create Index pattern을 선택합니다. 

이제 Index Pattern 이 생성되었습니다. 

 

5. 데이터 확인하기 

좌측메뉴의 Discorver로 이동합니다. 

Discover 의 우상단에 달력버튼을 통하여 시간 데이터값을 입력해주면, 데이터를 확인해 볼 수 있습니다. 

 

6. elasticsearch로 호출 

브라우저를 이용해서 데이터를 확인 해 보는 방법도 있습니다. 

url: http://ip:port/[index-name]/_search