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, (select json_build_object('region_id', region_id, 'region_name', region_name, 'region_name_kr', region_name_kr, 'region_type', region_type, 'region_code', region_code)::jsonb from expedia_region_union where region_id = c.region_id ) as region_json::text from expedia_airport_more c"
tags => ["expedia_airport_more"]
}
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, region_name_full_kr, continent::text, country::text from expedia_region_union order by region_id asc limit 100"
tags => ["expedia_region_union"]
}
stdin {
codec => plain { charset => "UTF-8"}
}
}
filter {
if "expedia_region_union" in [tags] {
ruby {
code => "
require 'json'
continent_json = JSON.parse(event.get('continent'.to_s))
event.set('continent', continent_json)
"
}
clone {
clones => ['test-a', 'test-b'] }
if [type] == 'test-a' {
prune {
whitelist_names => ["region_id", "region_name_kr"]
}
mutate {
add_field => {"[@metadata][type]" => "test-a"}
}
} else {
prune {
whitelist_names => ["region_id", "region_name_full_kr"]
}
mutate {
add_field => {"[@metadata][type]" => "test-b"}
}
}
} else {
ruby {
code => "require 'json'
begin
region_json_json = JSON.parse(event.get('region_json') || '{}')
event.set('region_json', region_json_json)
rescue Exception => e
event.tag('you have to check region_json sql.')
end
"
}
prune {
whitelist_names => ["region_id", "region_json"]
}
mutate {
add_field => {"[@metadata][type]" => "test-c"}
}
}
}
output {
stdout { codec => rubydebug }
if [@metadata][type] == 'test-c' {
elasticsearch {
hosts => ["localhost:9200"]
index => "test-c"
doc_as_upsert => true
action => "update"
document_id => "%{region_id}"
}
}
else if [@metadata][type] == 'test-a' {
elasticsearch {
hosts => [ "localhost:9200" ]
index => "test-a"
doc_as_upsert => true
action => "update"
document_id => "%{region_id}"
}
} else {
elasticsearch {
hosts => [ "localhost:9200" ]
index => "test-b"
doc_as_upsert => true
action => "update"
document_id => "%{region_id}"
}
}
}