logstash + amazon_es + jdbc: how to update and insert data

## logstash의 config 설정 파일

input { 
      jdbc{ 
           jdbc_driver_library => "/usr/share/java/mysql-connector-java-5.1.47/mysql-connector-java-5.1.47.jar" 
   
          jdbc_driver_class => "com.mysql.jdbc.Driver" 
   
          jdbc_connection_string => ###
   
          jdbc_user =>"admin" 
   
          jdbc_password => ###
   
          jdbc_paging_enabled => true 
           tracking_column => "contents_code" 
           use_column_value => true 
           tracking_column_type => "numeric" 
           schedule => "*/1 * * * *"  # 1분마다
   
   
          statement => "SELECT * FROM tbl_contents_master 
          WHERE mdate > (SELECT DATE_SUB(NOW(), INTERVAL 60 MINUTE))" 
      } 
  } 
  filter{ 
   
   fingerprint { 
          key => "1234ABCD" 
          method => "SHA256" 
          source => ["contents_code"] 
          target => "[@metadata][generated_id]" 
          concatenate_sources => true 
      } 
  } 
  output { 
      amazon_es { 
           hosts => ###
           region => "ap-northeast-2" 
           aws_access_key_id => ###
           aws_secret_access_key => ###
          index =>"new" 
          document_id => "%{[@metadata][generated_id]}" 
          doc_as_upsert => true 
          action => "update" 
   } 
   
      stdout{ 
          codec => rubydebug{} 
      } 
  }

 

1. logstash 1분마다 갱신되도록 : schedule => "*/1 * * * *"  # 1분마다

2. 현재시간과 비교해서 1분 전 안에 들어온 데이터를 전부 삽입

: "SELECT * FROM tbl_contents_master WHERE mdate > (SELECT DATE_SUB(NOW(), INTERVAL 60 MINUTE))" 

3. contents_code 를 기준으로 같은 코드면 덮어쓰게 함 : source => ["contents_code"] 

4. 새로운 데이터가 삽입되면 반영 : doc_as_upsert => true 

5. <update!!> output에 document_type => "_doc"