Logstash의 활용 - 새롭게 등록되거나 수정된 데이터만 등록

지난 시간에 작업했던 config는 testtable에 있는 모든 데이터를 매번 불러와서 그걸 또 매번 엘라스틱서치의 test인덱스에 넣도록 설정돼 있었다. 이렇게하면 당연히 비효율적일 수 밖에 없다. 이번엔 처음 가져올땐 전체를 가져오지만 그 이후에는 새롭게 등록되거나 수정된 데이터들만 가져와서 등록돼도록 config를 수정해보자.

 

/usr/share/logstash/data/test.conf

input {

  jdbc {

    jdbc_driver_library=>"/usr/share/logstash/bin/mysql-connector-j-8.0.33.jar"

    jdbc_driver_class=>"com.mysql.jdbc.Driver"

    jdbc_validate_connection=>true

    jdbc_connection_string=>"jdbc:mysql://localhost:3306/mydb?useSSL=false"

    jdbc_user=>"****"

    jdbc_password=>"**********"

    record_last_run => true

    clean_run => true

    tracking_column_type => "numeric"

    tracking_column => "unix_ts_in_secs"

    use_column_value => true

    schedule => "10 * * * * *"

    statement=>"SELECT *, UNIX_TIMESTAMP(edit_date) AS unix_ts_in_secs FROM testtable WHERE (UNIX_TIMESTAMP(edit_date) > :sql_last_value AND edit_date < NOW()) ORDER BY edit_date ASC"}

    }

filter {

  mutate {

    remove_field => ["@version", "testid", "testpoint", "unix_ts_in_secs"]

  }

}

output {

  elasticsearch {

    hosts=>["localhost:9200"]

    index=>"test"

    document_id=>"%{num}"

    user => "elastic"

    password => "**********"

    }

    stdout {}

  }

지난번 보다 뭔가 더 추가가 됐다. 하나씩 살펴보자.

 

    record_last_run => true

    clean_run => true

    tracking_column_type => "numeric"

    tracking_column => "unix_ts_in_secs"

    use_column_value => true

여기서 중요한것은 tracking_column이다. 어떤 컬럼이 기준이 되느냐 하는 건데 아래에 있는 쿼리문에도 들어가 있다. 그리고 그 컬럼은 숫자 타입이라는 것이다. 쿼리를 살펴보자.

 

statement=>"SELECT *, UNIX_TIMESTAMP(edit_date) AS unix_ts_in_secs FROM testtable WHERE (UNIX_TIMESTAMP(edit_date) > :sql_last_value AND edit_date < NOW()) ORDER BY edit_date ASC"}

테이블에 등록하거나 수정할때 항상edit_date의 값을 업데이트 해준다. 그리고 그 값을 "unix_ts_in_secs"에 넣어주었다. 그리고 그 저장된 값을 불러올때는 ":sql_last_value"로 불러온다.

 

쿼리문을 보면 알겠지만 마지막 edit_date의 시간 이후로 등록되거나 수정된 데이터들을 불러오라는 뜻이다.

 

filter {

  mutate {

    remove_field => ["@version", "testid", "testpoint", "unix_ts_in_secs"]

  }

}

filter는 디비에서 인덱스로 데이터를 넘길때 필터링 할것들을 지정할때 사용한다. 여러가지 넣을 수가 있는데 이번엔 쿼리에서 나오는 컬럼들중에서 인덱스에 넣을 필요가 없는 컬럼들을 지정해서 그 컬럼들은 들어가지 않도록 해 주었다. 물론 이 부분은 없어도 된다.

 

filter에 들어갈 수 있는 옵션을 더 알고 싶다면 다음 블로그를 참조하자.

 

https://gem1n1.tistory.com/185

[ELK Logstash] config 파일 작성 예시 - input, filter, output

https://logz.io/blog/logstash-mutate-filter/

Using the Mutate Filter in Logstash | Logz.io

 

물론 이렇게 안하고 php로 파일 하나를 만들어도 된다. 그걸 크론으로 매일 돌려도 되는거고. 하지만 이런 기능도 있으니 한번 써보는 것도 나쁘지 않은 것 같아 소개했다.