# Db streaming from last input

**URL:** <https://discuss.elastic.co/t/db-streaming-from-last-input/62905>\
**Category:** Logstash\
**Created:** [October 13, 2016, 7:17am UTC](https://discuss.elastic.co/t/db-streaming-from-last-input/62905 "2016-10-13T07:17:59Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![higee](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/higee/32/22343_2.png) [@higee](https://discuss.elastic.co/u/higee)\
**Post date:** [October 13, 2016, 7:17am UTC](https://discuss.elastic.co/t/db-streaming-from-last-input/62905/1 "2016-10-13T07:17:59Z")

</div>

Hi, I'm using logstash to extract data from database and send to elasticsearch and to kibana.  
What I want to do is I want elasticsesarch to store past data and just get newly inserted data to my oracle database. For instance, if data [No.1] ~ [No.10] have already been inserted to Elasticsearch, I want logstash to just read data from [No.11] ~ [No. most recent] and store them to Elasticsearch. Since I couldn't figure that out, I alternatively used duplicate filter, please refer to the code below to see how I did.

1. read every data till now:

> TO\_CHAR(DATE, 'yyyy-mm-dd HH24:MI') \< TO\_CHAR(SYSDATE, 'yyyy-mm-dd HH24:MI')

1. remove duplicates

> filter {  
> mutate {  
> add\_field =\> {  
> "[@metadata][document\_id]" =\> "%{IDX}"  
> }  
> }  
> }

But I guess this is waste of resource since logstash will read the whole data first and then remove the duplicates. Is there a better way to improve my code?

Thanks

Best

gee

* * *

P.S.

My logstash.conf file is as follows:

> input {  
> jdbc {  
> jdbc\_validate\_connection =\>  
> jdbc\_connection\_string =\>  
> jdbc\_user =\>  
> jdbc\_password =\>  
> jdbc\_driver\_library =\>  
> jdbc\_driver\_class =\>  
> statement =\>  
> "  
> SELECT IDX, DATE, PRICE, PROFIT  
> FROM TABLE\_NAME  
> WHERE  
> TO\_CHAR(DATE, 'yyyy-mm-dd HH24:MI') \< TO\_CHAR(SYSDATE, 'yyyy-mm-dd HH24:MI')  
> "  
> }  
> }  
> filter {  
> mutate {  
> add\_field =\> {  
> "[@metadata][document\_id]" =\> "%{IDX}"  
> }  
> }  
> }  
> output {  
> elasticsearch {  
> index =\>  
> hosts =\>  
> user =\>  
> password =\>  
> document\_id =\> "%{[@metadata][document\_id]}"  
> }  
> }

I skipped all the sensitive part, (e.g. user, password..)

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [October 13, 2016, 9:59am UTC](https://discuss.elastic.co/t/db-streaming-from-last-input/62905/2 "2016-10-13T09:59:56Z")

</div>

The jdbc input does this for you. Set its `tracking_column` option to the name of the timestamp column and enable the feature with `use_column_value`. Then the plugin will maintain a query parameter with the timestamp (or whatever the `tracking_column` column contains) of the previous run, allowing you to select entries newer than that. See the plugin documentation for an example.

---

<div class="post-metadata">

**Author:** ![higee](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/higee/32/22343_2.png) [@higee](https://discuss.elastic.co/u/higee)\
**Post date:** [October 16, 2016, 5:57pm UTC](https://discuss.elastic.co/t/db-streaming-from-last-input/62905/3 "2016-10-16T17:57:15Z")

</div>

thx for the reply!  
I'll check it out right away and come back to leave a comment

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [July 6, 2017, 4:34am UTC](https://discuss.elastic.co/t/db-streaming-from-last-input/62905/4 "2017-07-06T04:34:02Z")

</div>