JDBC Sqlserver to elastic search : one input to different output? I have a sqlserver2016 database that I want to index in elasticsearch using logstash.

Here is my logstash config file which somewhat works :

input {
jdbc {
jdbc_driver_library =&gt; "C:\elastic\Microsoft-JDBC-Driver-6.0-for-SQL-Server\sqljdbc_6.0\enu\jre8\sqljdbc42.jar"
jdbc_driver_class =&gt; "com.microsoft.sqlserver.jdbc.SQLServerDriver"
jdbc_connection_string =&gt; "jdbc:sqlserver://10.11.12.13:1433;databaseName=testdb1;integratedSecurity=false;user=ElasticExtractor;password=flyingweisels;"
jdbc_user =&gt; "ElasticExtractor"
jdbc_password =&gt; "flyingweisels"
statement =&gt; "select top 150000 Item_ID itemid,merchant_id merchantid,modelnumber,language_id from items order by Item_Id desc"
}
}
output {
elasticsearch {
hosts =&gt; "localhost:9200"
index =&gt; "testdata"
document_type =&gt; "testtype"
document_id =&gt; "%{itemid}"
}
}


So what this file is supposed to do, as configured is insert 150k items in elasticSearch. Somewhat it only imports about a third of that, such as 62 382 in this case. If I try to insert 50k it only inserts about 20k.
Is there an obvious reason why it would do that ?

Here is the current execution log :

[2017-09-01T08:16:31,923][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=&gt;{:removed=&gt;[], :added=&gt;[http://localhost:9200/]}}
[2017-09-01T08:16:31,927][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=&gt;http://localhost:9200/, :path=&gt;"/"}
[2017-09-01T08:16:32,006][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=&gt;"http://localhost:9200/"}
[2017-09-01T08:16:32,007][INFO ][logstash.outputs.elasticsearch] Using mapping template from {:path=&gt;nil}
[2017-09-01T08:16:32,042][INFO ][logstash.outputs.elasticsearch] Attempting to install template {:manage_template=&gt;{"template"=&gt;"logstash-*", "version"=&gt;50001, "settings"=&gt;{"index.refresh_interval"=&gt;"5s"}, "mappings"=&gt;{"_default_"=&gt;{"_all"=&gt;{"enabled"=&gt;true, "norms"=&gt;false}, "dynamic_templates"=&gt;[{"message_field"=&gt;{"path_match"=&gt;"message", "match_mapping_type"=&gt;"string", "mapping"=&gt;{"type"=&gt;"text", "norms"=&gt;false}}}, {"string_fields"=&gt;{"match"=&gt;"*", "match_mapping_type"=&gt;"string", "mapping"=&gt;{"type"=&gt;"text", "norms"=&gt;false, "fields"=&gt;{"keyword"=&gt;{"type"=&gt;"keyword", "ignore_above"=&gt;256}}}}}], "properties"=&gt;{"@timestamp"=&gt;{"type"=&gt;"date", "include_in_all"=&gt;false}, "@version"=&gt;{"type"=&gt;"keyword", "include_in_all"=&gt;false}, "geoip"=&gt;{"dynamic"=&gt;true, "properties"=&gt;{"ip"=&gt;{"type"=&gt;"ip"}, "location"=&gt;{"type"=&gt;"geo_point"}, "latitude"=&gt;{"type"=&gt;"half_float"}, "longitude"=&gt;{"type"=&gt;"half_float"}}}}}}}}
[2017-09-01T08:16:32,050][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=&gt;"LogStash::Outputs::ElasticSearch", :hosts=&gt;["//localhost:9200"]}
[2017-09-01T08:16:32,053][INFO ][logstash.pipeline        ] Starting pipeline {"id"=&gt;"main", "pipeline.workers"=&gt;4, "pipeline.batch.size"=&gt;125, "pipeline.batch.delay"=&gt;5, "pipeline.max_inflight"=&gt;500}
[2017-09-01T08:16:32,219][INFO ][logstash.pipeline        ] Pipeline main started
[2017-09-01T08:16:32,313][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=&gt;9600}
[2017-09-01T08:16:32,643][INFO ][logstash.inputs.jdbc     ] (0.050000s) select top 150000 Item_ID itemid,merchant_id merchantid,modelnumber,language from items order by Item_Id desc
[2017-09-01T08:16:49,805][WARN ][logstash.agent           ] stopping pipeline {:id=&gt;"main"}


Second thing is, let’s say I want to insert a row from SQL server that comes from this input, what plugin can I use so that if the row has a specific “merchant_id” it goes in an elastic TYPE named with that ID. Also, if it has a specific “language” it goes in an elastic INDEX with that language as the name.
Can that be done ?
Should I simply create multiple Logstash config files, one for each of those tasks ?
So I figured out what I was doing wrong, there were 2 different problems.
1. Wrong inserted row count
This was caused by the SQL query I was using. Elastic wants unique document_id values and the left join I was using caused multiple rows to be returned with the same values. What elastic does in such case is it overwrites the existing row with the new values.
2. Assign rows to different indexes and types
I had tried using the same syntax as in the document_id which ended up looking like this:
index =&gt; &quot;%{Language}&quot;
document_type =&gt; &quot;%{MerchantID}&quot;

This would result in rows getting inserted with in an index called %{Language} which was obviously not the desired result.
The problem was simple: ELASTIC DOES NOT LIKE UPPERCASE LETTERS.
So my final configuration file ended up looking like this:
input {
jdbc {
jdbc_driver_library =&gt; &quot;C:\elastic\Microsoft-JDBC-Driver-6.0-for-SQL-Server\sqljdbc_6.0\enu\jre8\sqljdbc42.jar&quot;
jdbc_driver_class =&gt; &quot;com.microsoft.sqlserver.jdbc.SQLServerDriver&quot;
jdbc_connection_string =&gt; &quot;jdbc:sqlserver://10.11.12.13:1433;databaseName=testdb1;integratedSecurity=false;user=ElasticExtractor;password=flyingweisels;&quot;
jdbc_user =&gt; &quot;ElasticExtractor&quot;
jdbc_password =&gt; &quot;flyingweisels&quot;
statement =&gt; &quot;select top 50000 id itemid,item_id itemid,Merchant_ID merchantid,model,case when (ID%2=0) then 'germanindex' else 'englishindex' end language from items order by id desc&quot;
}
}
output {
elasticsearch {
hosts =&gt; &quot;localhost:9200&quot;
index =&gt; &quot;%{language}&quot;
document_type =&gt; &quot;%{merchantid}&quot;
document_id =&gt; &quot;%{itemid}&quot;
}
}

It works great !