# Jdbc plugin ships Repeated data from mongodb **URL:** \ **Category:** Logstash\ **Created:** [July 11, 2019, 6:38am UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914 "2019-07-11T06:38:40Z")\ **Posts on this page:** 17\ **Page:** 1
**Author:** ![chandu5565](https://avatars.discourse-cdn.com/v4/letter/c/c57346/32.png) [@chandu5565](https://discuss.elastic.co/u/chandu5565)\ **Post date:** [July 11, 2019, 6:38am UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/1 "2019-07-11T06:38:40Z")
Hi, I am using logstash JDBC plugin to ship data from Mongo DB to elasticsearch. The data stored in the Mongo DB is in JSON documents. When logstash try to fetch the data every 1 min its getting repeated data. Total documents is DB is 20378 but my logstash is shipping all 20378 documents on every minute it is scheduled. Need immediate help. Below is the logstash configuration file. \ input { jdbc { jdbc\_driver\_library =\>"C:/Users/home/Downloads/Backup/Downloads/mongo-java-driver-3.4.2.jar,C:/Users/home/Downloads/Backup/Downloads/mongojdbc1.2.jar" jdbc\_driver\_class =\> "com.dbschema.MongoJdbcDriver" jdbc\_connection\_string =\> "jdbc:mongodb://\*\*\*:\*\*\*/\*\*\*\*\*_" jdbc\_user =\> "admin" schedule =\> "_/1 \* \* \* \*" statement =\> "db.DeviceServiceData.find()" last\_run\_metadata\_path =\> "C:/Users/home/.logstash\_jdbc\_last\_run" } } output { elasticsearch { hosts =\> "localhost:9200" index =\> "myindex" ``` } stdout { codec => rubydebug } ``` } \ By running the above logstash configuration file am able to receive the data but it is repeated. every minute logstash goes and fetches the complete data in DB. I need only the updated. Below are the logs of logstash. \ [2019-07-11T06:07:01,719][INFO][logstash.runner] Starting Logstash {"logstash.version"=\>"7.1.1"} [2019-07-11T06:07:10,202][INFO][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=\>{:removed=\>, :added=\>[[http://localhost:9200/](http://localhost:9200/)]}} [2019-07-11T06:07:10,437][WARN][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=\>"[http://localhost:9200/](http://localhost:9200/)"} [2019-07-11T06:07:10,496][INFO][logstash.outputs.elasticsearch] ES Output version determined {:es\_version=\>7} [2019-07-11T06:07:10,500][WARN][logstash.outputs.elasticsearch] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document \_type {:es\_version=\>7} [2019-07-11T06:07:10,527][INFO][logstash.outputs.elasticsearch] New Elasticsearch output {:class=\>"LogStash::Outputs::ElasticSearch", :hosts=\>["[//localhost:9200](https://localhost:9200)"]} [2019-07-11T06:07:10,541][INFO][logstash.outputs.elasticsearch] Using default mapping template [2019-07-11T06:07:10,563][INFO][logstash.javapipeline] Starting pipeline {:pipeline\_id=\>"main", "pipeline.workers"=\>4, "pipeline.batch.size"=\>125, "pipeline.batch.delay"=\>5, "pipeline.max\_inflight"=\>500, :thread=\>"#\"} [2019-07-11T06:07:10,748][INFO][logstash.outputs.elasticsearch] Attempting to install template {:manage\_template=\>{"index\_patterns"=\>"logstash-_", "version"=\>60001, "settings"=\>{"index.refresh\_interval"=\>"5s", "number\_of\_shards"=\>1}, "mappings"=\>{"dynamic\_templates"=\>[{"message\_field"=\>{"path\_match"=\>"message", "match\_mapping\_type"=\>"string", "mapping"=\>{"type"=\>"text", "norms"=\>false}}}, {"string\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"string", "mapping"=\>{"type"=\>"text", "norms"=\>false, "fields"=\>{"keyword"=\>{"type"=\>"keyword", "ignore\_above"=\>256}}}}}], "properties"=\>{"@timestamp"=\>{"type"=\>"date"}, "@version"=\>{"type"=\>"keyword"}, "geoip"=\>{"dynamic"=\>true, "properties"=\>{"ip"=\>{"type"=\>"ip"}, "location"=\>{"type"=\>"geo\_point"}, "latitude"=\>{"type"=\>"half\_float"}, "longitude"=\>{"type"=\>"half\_float"}}}}}}} [2019-07-11T06:07:10,879][INFO][logstash.javapipeline] Pipeline started {"pipeline.id"=\>"main"} [2019-07-11T06:07:11,038][INFO][logstash.agent] Pipelines running {:count=\>1, :running\_pipelines=\>[:main], :non\_running\_pipelines=\>} [2019-07-11T06:07:12,075][INFO][logstash.agent] Successfully started Logstash API endpoint {:port=\>9600} [2019-07-11T06:08:01,141][INFO][org.mongodb.driver.cluster] Cluster created with settings {hosts=_**:\*\*\*\*\*], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500} [2019-07-11T06:08:01,227][INFO][org.mongodb.driver.connection] Opened connection [connectionId{localValue:1, serverValue:2954530}] to \*: [2019-07-11T06:08:01,230][INFO][org.mongodb.driver.cluster] Monitor thread successfully connected to server with description ServerDescription{address=:**__, type=STANDALONE, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 6, 4]}, minWireVersion=0, maxWireVersion=6, maxDocumentSize=16777216, roundTripTimeNanos=1218900} [2019-07-11T06:08:01,664][INFO][org.mongodb.driver.connection] Opened connection [connectionId{localValue:2, serverValue:2954531}] to_\*\*\*\*\*:\*\*\*\*\* [2019-07-11T06:08:01,901][INFO][logstash.inputs.jdbc] (0.496248s) db.DeviceServiceData.find() [2019-07-11T06:09:00,260][INFO][org.mongodb.driver.cluster] Cluster created with settings {hosts=[_ **:** _], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500} [2019-07-11T06:09:00,270][INFO][org.mongodb.driver.connection] Opened connection [connectionId{localValue:3, serverValue:2954584}] to _ **:** _ [2019-07-11T06:09:00,270][INFO][org.mongodb.driver.cluster] Monitor thread successfully connected to server with description ServerDescription{address=_**:\*\*\*\*, type=STANDALONE, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 6, 4]}, minWireVersion=0, maxWireVersion=6, maxDocumentSize=16777216, roundTripTimeNanos=926300} [2019-07-11T06:09:00,313][INFO][org.mongodb.driver.connection] Opened connection [connectionId{localValue:4, serverValue:2954585}] to : [2019-07-11T06:09:00,334][INFO][logstash.inputs.jdbc] (0.034389s) db.DeviceServiceData.find() [2019-07-11T06:10:00,476][INFO][org.mongodb.driver.cluster] Cluster created with settings {hosts=[**_:**], mode=SINGLE, requiredClusterType=UNKNOWN, serverSelectionTimeout='30000 ms', maxWaitQueueSize=500} [2019-07-11T06:10:00,485][INFO][org.mongodb.driver.connection] Opened connection [connectionId{localValue:5, serverValue:2954640}] to : [2019-07-11T06:10:00,486][INFO][org.mongodb.driver.cluster] Monitor thread successfully connected to server with description ServerDescription{address=**:\*\*\*\*, type=STANDALONE, state=CONNECTED, ok=true, version=ServerVersion{versionList=[3, 6, 4]}, minWireVersion=0, maxWireVersion=6, maxDocumentSize=16777216, roundTripTimeNanos=836600} [2019-07-11T06:10:00,504][INFO][org.mongodb.driver.connection] Opened connection [connectionId{localValue:6, serverValue:2954641}] to **:** [2019-07-11T06:10:00,521][INFO][logstash.inputs.jdbc] (0.025582s) db.DeviceServiceData.find() \ ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [July 11, 2019, 1:33pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/2 "2019-07-11T13:33:27Z")
If you want the jdbc to maintain state then the query has to depend on sql\_last\_value. The [documentation](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-jdbc.html#_predefined_parameters) has an example using a simple SQL statement. ---
**Author:** ![chandu5565](https://avatars.discourse-cdn.com/v4/letter/c/c57346/32.png) [@chandu5565](https://discuss.elastic.co/u/chandu5565)\ **Post date:** [July 11, 2019, 2:22pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/3 "2019-07-11T14:22:21Z")
Thanks for reply Badger, I have gone through that documentation and I updated the last\_run\_metadata\_path to /.logstash\_jdbc\_last\_run that didn't solve my problem. Again am getting duplicate data and it's repeating. Toatl documents I have in my db is 20000 but every minute my logstash is fetching 20000. ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [July 11, 2019, 2:23pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/4 "2019-07-11T14:23:44Z")
Have you modified your stored procedure to filter the result set based on the value of sql\_last\_value? ---
**Author:** ![chandu5565](https://avatars.discourse-cdn.com/v4/letter/c/c57346/32.png) [@chandu5565](https://discuss.elastic.co/u/chandu5565)\ **Post date:** [July 11, 2019, 2:32pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/5 "2019-07-11T14:32:27Z")
Yes I have tried changing statement like below Statement =\> from \* db.subscriber\_inventory.find() where sequenceid \> sql\_last\_value Even that didn't work ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [July 11, 2019, 2:36pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/6 "2019-07-11T14:36:56Z")
I do not do SQL beyond the most basic SELECTs, so I am unable to assist further. ---
**Author:** ![chandu5565](https://avatars.discourse-cdn.com/v4/letter/c/c57346/32.png) [@chandu5565](https://discuss.elastic.co/u/chandu5565)\ **Post date:** [July 11, 2019, 2:39pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/7 "2019-07-11T14:39:39Z")
My database is mongodb if can help me with this it would be great am really helpless running out of storage and many issues because of this repeated data. Thanks for your help. ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [July 11, 2019, 3:31pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/8 "2019-07-11T15:31:47Z")
As I said, I cannot help with the SQL. It would be possible to use a fingerprint filter to create a document\_id for the elasticsearch output. That would result in you overwriting the documents each time rather than creating new documents. Much less efficient than filtering in the SQL but would resolve part of the problem. ---
**Author:** ![chandu5565](https://avatars.discourse-cdn.com/v4/letter/c/c57346/32.png) [@chandu5565](https://discuss.elastic.co/u/chandu5565)\ **Post date:** [July 18, 2019, 2:00pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/9 "2019-07-18T14:00:09Z")
Hi Badger, I have added documentid parameter in output and assigned it to unique value of my document like below in my configuration file. output { elasticsearch { hosts =\> ["localhost:9200"] index =\> "commands" document\_id =\> "%{messageid}" } stdout { codec =\> rubydebug } } Now what happend is 1. It stop taking multiple documents but is every time when my logstash runs it is overwriting the existing documents which has the same unique document id which i had given. Time stamp is getting update every time as new timestamp. If it is like this i cant get the data for last 7 days or so. 2. When ever my logstash configuration runs as those or the duplicate documents my document count is not increasing but the storage size of the index is increasing. ---
**Author:** ![chandu5565](https://avatars.discourse-cdn.com/v4/letter/c/c57346/32.png) [@chandu5565](https://discuss.elastic.co/u/chandu5565)\ **Post date:** [July 18, 2019, 2:04pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/10 "2019-07-18T14:04:34Z")
And also can you give the logstash configuration by adding fingerprint filter which you had suggested by making necessary changes to my configuration which i given. ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [July 18, 2019, 2:21pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/11 "2019-07-18T14:21:31Z")
I have no idea what your data looks like so I cannot suggest how to configure the fingerprint filter. I really think you should focus on modifying the stored procedure so that it accepts :sql\_last\_value as a parameter and filters out older records. ---
**Author:** ![chandu5565](https://avatars.discourse-cdn.com/v4/letter/c/c57346/32.png) [@chandu5565](https://discuss.elastic.co/u/chandu5565)\ **Post date:** [July 18, 2019, 2:31pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/12 "2019-07-18T14:31:33Z")
I had setup logstash\_last\_run\_path and it is getting updated with the lastrun timestamp as below --- 2019-07-18 06:10:01.992880000 Z But in my Document timestamp filed is in ISODATE format like below 2019-05-27T20:07:45.486Z Can you help me in changing the ISODATE format to format of my timestamp field. ---
**Author:** ![chandu5565](https://avatars.discourse-cdn.com/v4/letter/c/c57346/32.png) [@chandu5565](https://discuss.elastic.co/u/chandu5565)\ **Post date:** [July 18, 2019, 2:33pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/13 "2019-07-18T14:33:41Z")
Can i change the default logstash timestamp to ISODATE format. If yes how can I change. ---
**Author:** ![Badger](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/badger/32/25190_2.png) [@Badger](https://discuss.elastic.co/u/Badger)\ **Post date:** [July 18, 2019, 3:11pm UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/14 "2019-07-18T15:11:28Z")
> [@chandu5565](#): > > Can you help me in changing the ISODATE format to format of my timestamp field. I do not know how to do that. ---
**Author:** ![chandu5565](https://avatars.discourse-cdn.com/v4/letter/c/c57346/32.png) [@chandu5565](https://discuss.elastic.co/u/chandu5565)\ **Post date:** [July 25, 2019, 9:33am UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/15 "2019-07-25T09:33:05Z")
The timestamp stored by the logstash\_jdbc\_last\_run is not an ISODATE format can anyone help me in changing the format of logstash\_jdbc\_last\_run to ISODATE format. ---
**Author:** ![chandu5565](https://avatars.discourse-cdn.com/v4/letter/c/c57346/32.png) [@chandu5565](https://discuss.elastic.co/u/chandu5565)\ **Post date:** [August 22, 2019, 7:16am UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/16 "2019-08-22T07:16:43Z")
Hi Badger, Thanks for your help. I found a solution for this repeated data. Am posting so it may be helpful to someone. I have changed the statement and written a conditional statement and I had used fingerprint to add new field with the value of timestamp in my data which is unique and incremental for every document. ```auto input { jdbc { jdbc_driver_library =>"C:/Users/home/Downloads/Backup/Downloads/mongo-java-driver-3.4.2.jar,C:/Users/home/Downloads/Backup/Downloads/mongojdbc1.2.jar" jdbc_driver_class => "com.dbschema.MongoJdbcDriver" jdbc_connection_string => "jdbc:mongodb:// ***:*** / ***** " jdbc_user => "admin" schedule => "*/60 * * * *" statement => "db.databasename.find({ timestampfield: { $gte: (:sql_last_value)}})" last_run_metadata_path => "C:/Users/home/.logstash_jdbc_last_run" } } filter { fingerprint { add_field => { "fingerprint" => "%{timestampfield}" } } } output { elasticsearch { hosts => "localhost:9200" index => "myindex" } stdout { codec => rubydebug } } ``` ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [September 19, 2019, 7:16am UTC](https://discuss.elastic.co/t/jdbc-plugin-ships-repeated-data-from-mongodb/189914/17 "2019-09-19T07:16:44Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.