logstash clone 설정

하나의 인풋. 하나의 포트에서 multiple output... same host에 multiple 한 index output 설정

 

정확히 말하면 하나의 테이블에서 2개의 인덱스 동기화

또 다른 테이블도 가져와야하긴 함.

 

input {
    jdbc {
        jdbc_connection_string => "jdbc:postgresql://localhost:5432/atlasdb?useTimezone=true&useLegacyDatetimeCode=false&serverTimezone=UTC&useSSL=false&useUnicode=true&characterEncoding=utf8"
        jdbc_user => "atlas"
        jdbc_password => "atlas"
        jdbc_validate_connection => true
        jdbc_driver_library => "/lib/postgres-42-test.jar"
        jdbc_driver_class => "org.postgresql.Driver"
        schedule => "* * * * *"
        statement => "SELECT region_id, region_name, region_name_full, region_name_kr, continent::text, country::text from expedia_region_union order by region_id asc limit 100"
   }

   stdin {
        codec => plain { charset => "UTF-8"}
   }

}
filter {
        ruby {
                code => "
                        require 'json'
                        continent_json = JSON.parse(event.get('continent').to_s)
                        event.set('continent', continent_json)
                        country_json = JSON.parse(event.get('country').to_s)
                        event.set('country', country_json)
                        "
                }
        date {
                match => ['time', 'UNIX']
        }
        clone {
                clones => ['2020-05-21-thu', '2020-06-17-wed']
        }
        if [type] == '2020-06-17-wed' {
                prune {
                        whitelist_names => ["region_id", "region_name_full"]
                }
                mutate {
                        add_field => {"[@metadata][type]" => "only_region_name_full"}
                }
        }
        else if [type] == '2020-05-21-thu' {
                prune {
                        whitelist_names => ["region_id", "region_name"]
                }
                mutate {
                        add_field => {"[@metadata][type]" => "only_region_name"}
                }
        }
}

output {
        stdout { codec => rubydebug }

        if [@metadata][type] == 'only_region_name_full' {
                elasticsearch {
                        hosts => ["localhost:9200"]
                        index => "2020-06-17-wed"
                        doc_as_upsert => true
                        action => "update"
                        document_id => "%{region_id}"
                }
        }
        else {
                elasticsearch {
                        hosts => [ "localhost:9200" ]
                        index => "2020-05-21-thu"
                        doc_as_upsert => true
                        action => "update"
                        document_id => "%{region_id}"
                }
        }

}