[ElasticSearch] Rdbms와 ElasticSearch 연동하기

1. Rdbms와 ElasticSearch 연동 조건 

elasticSearch를 이용해서 검색엔진을 개발하려면 그 서비스에서 사용하고 있는 데이터베이스와 연동이 필수적이다. 

원할한 검색 엔진을 만들기 위해서는 JDBC 입력 플로그인에 있는 Logstash를 이용하여 ElasticSearch와 MYSQL 동기화를 유지해야한다.

정확히 작동하려면 RDBMS에 몇가지 조건이 필요하다.

 

조건 1.  MySQL 문서가 ElasticSearch로 작성될 때, ElasticSearch의 "_id" 필드는 MySql "id" 필드로 설정되어야한다. 이렇게 설정하면 MySQL 레코드와 ElasticSearch 문서간에 직접 매핑을 제공해준다.

레코드가 MySQL에서 업데이트 되는 경우, 연결된 문서 전체가 ElasticSearch에서 덮어쓰기된다. (효율성은 떨어지는 작업이다.) 

 

조건 2. MySQL에 레코드가 삽입되거나 업데이트되면, 그 레코드는 업데이트 또는 삽입시간을 포함하는 필드(칼럼)를 가져야한다. (이는 서비스의 마케팅이나 데이터적 차원에서도 긍정적인 기능을 한다.) LogStash는 이 필드(칼럼)들이 폴링 루프의 마지막 반복 계산 이래 수정되거나 삽입된 문서만 요청할 수 있도록 하는데 이용한다. LogStash 폴링 루프의 이전 반복 계산에서 받았던 마지막 기록보다 새로운 업데이트나 삽입이 있는 레코드만 요청해야한다는 것을 알고있다. 

 

2. LOGSTASH 동기화 코드 

input {
  jdbc {
    jdbc_driver_library => "<path>/mysql-connector-java-8.0.16.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://<MySQL host>:3306/es_db"
    jdbc_user => <my username>
    jdbc_password => <my password>
    jdbc_paging_enabled => true
    tracking_column => "unix_ts_in_secs"
    use_column_value => true
    tracking_column_type => "numeric"
    schedule => "*/5 * * * * *"
    statement => "데이터 가져오는 쿼리문 (select 문)"
  }
}
filter {
  mutate {
    copy => { "id" => "[@metadata][_id]"}
    remove_field => ["id", "@version", "unix_ts_in_secs"]
  }
}
output {
  # stdout { codec =>  "rubydebug"}
  elasticsearch {
      index => "rdbms_sync_idx"
      document_id => "%{[@metadata][_id]}"
  }
}

 

- tracking_column: LogStash가 MYSQL로부터 읽은 마지막 문서를 추적하는데 사용된다. 

 

- unix_ts_in_secs: SELECT 문이 생성하는 필드이고, 표준 Unix 타임스탬프로서 "modification_time" 을 포함한다. tracking_column에 의해 참조된다. 일반 타임스탬프로 사용되기보다는 진행상황을 추적할 때 사용된다.

 

- sql_last_value: LogStash의 폴링 루프의 현재 반복 계산을 위한 시작점을 포함하는 기본 제공 매개변수이며, 위의 jdbc 입력 구성의 SELECT 문에서 참조된다. 이것은 "unix_tx_in_secs"의 가장 최근 값으로 설정되며, .logstash_jdbc_last_run으로부터 읽는다. 

LogStash의 폴링 루프에서 실행되는 MYSQL 쿼리가 반환하는 문서를 위한 시작점으로 사용된다. 쿼리에서 사용된다면, 구분지어줘야한다.

 

- schedule: 얼마나 자주 MySQL을 폴링해야하는지 지정한다. 

 

- filter: MySQL 레코드로부터 "id"의 값을 "_id"라고 하는 메타데이터 필드로 그냥 복사한다. 그리고 이것은 문서가 정확한 "_id" 값을 가지고 ElasticSearch로 작성되도록 하기 위해서 출력에서 참조된다. 메타데이터 필드를 사용하면 이 임시값이 새 필드로 생성되지 않도록 해준다. 

 

- output: 각 문서가 ElasticSearch로 작성되고, 우리가 필터 섹션에서 생성한 메타데이터 필드로부터 가져오는 "_id"에 할당되도록 지정한다. 디버깅의 도움으로 활성화될 수 있는 주석처리된 rubydebug 출력도 있다. 

 

- modification_time < NOW(): 데이터를 정확하게 가져오기에 필요한 필드이다. 

 

3. 삭제시

위 방법으로는 삭제시 elasticSearch가 적용되지 않는다. (실제 서비스에서 데이터가 삭제되어도 삭제하지 않는 경우가 많다. )

"is_deleted" 처럼 데이터의 상태를 나타내는 컬럼을 두고, LogStatsh를 통해 그 변경사항이 ElasticSearch에 적용되도록 만들어주면된다. 대신 MySQL 쿼리는 "is_deleted"가 참인 레코드/문서를 제외할 수 있도록 작성되어야한다. 

 

참고) https://www.elastic.co/kr/blog/how-to-keep-elasticsearch-synchronized-with-a-relational-database-using-logstash