# Duplication of data using the aggregate plugin

**URL:** <https://discuss.elastic.co/t/duplication-of-data-using-the-aggregate-plugin/139284>\
**Category:** Logstash\
**Created:** [July 10, 2018, 7:19am UTC](https://discuss.elastic.co/t/duplication-of-data-using-the-aggregate-plugin/139284 "2018-07-10T07:19:24Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![wm1994](https://avatars.discourse-cdn.com/v4/letter/w/e79b87/32.png) [@wm1994](https://discuss.elastic.co/u/wm1994)\
**Post date:** [July 10, 2018, 7:19am UTC](https://discuss.elastic.co/t/duplication-of-data-using-the-aggregate-plugin/139284/1 "2018-07-10T07:19:24Z")

</div>

The first time we used logstash to process data today, there was a problem of data duplication or loss.This is logstash's bug or I've got a mistake?  
And my logstash.conf is:

> input {  
> stdin {}  
> jdbc {  
> jdbc\_connection\_string =\> "jdbc:mysql://192.168.88.128:3306/gwbnsh\_common"  
> jdbc\_driver\_library =\> "/usr/local/mysql-connector-java-5.1.7-bin.jar"  
> jdbc\_driver\_class =\> "com.mysql.jdbc.Driver"  
> jdbc\_user =\> "root"  
> jdbc\_password =\> "123"  
> clean\_run =\> false  
> last\_run\_metadata\_path =\> "/usr/local/logstash-6.2.1/template/infos/system\_district"  
> record\_last\_run =\> true  
> use\_column\_value =\> true  
> tracking\_column =\> "id"  
> lowercase\_column\_names =\> false  
> jdbc\_paging\_enabled =\> true  
> jdbc\_page\_size =\> "50000"  
> #statement\_filepath =\> "/usr/local/logstash-6.2.1/bin/logstash\_mysql\_sync/mysql\_sync.sql"  
> statement =\> "SELECT  
> t1.ID id,  
> t1.`NAME` name,  
> t1.BOSS\_CODE bossCode,  
> t1.NAMESPACE\_CODE namespaceCode,  
> t1.CREATOR creator,  
> DATE\_FORMAT(t1.CREATED\_DATE, '%Y-%m-%d %H:%I:%S') createdDate,  
> t1.UPDATER updater,  
> DATE\_FORMAT(t1.UPDATED\_DATE, '%Y-%m-%d %H:%I:%S') updatedDate,  
> t1.IS\_DEL isDel,  
> t2.ID id\_s,  
> t2.`NAME` name\_s,  
> t2.REGION\_ID rid\_s,  
> t3.MANGER m\_s,  
> t2.REMARK r\_s,  
> t2.BOSS\_CODE b\_s,  
> t2.NAMESPACE\_CODE n\_s,  
> t2.CREATOR c\_s,  
> DATE\_FORMAT(t2.CREATED\_DATE, '%Y-%m-%d %H:%I:%S') cd\_s,  
> t2.UPDATER u\_s,  
> DATE\_FORMAT(t2.UPDATED\_DATE, '%Y-%m-%d %H:%I:%S') ud\_s,  
> t2.IS\_DEL i\_s,  
> t3.ID id\_f,  
> t3.`NAME` name\_f,  
> t3.REGION\_ID rid\_f,  
> t3.MANGER m\_f,  
> t3.REMARK r\_f,  
> t3.BOSS\_CODE b\_f,  
> t3.NAMESPACE\_CODE n\_f,  
> t3.CREATOR c\_f,  
> DATE\_FORMAT(t3.CREATED\_DATE, '%Y-%m-%d %H:%I:%S') cd\_f,  
> t3.UPDATER u\_f,  
> DATE\_FORMAT(t3.UPDATED\_DATE, '%Y-%m-%d %H:%I:%S') ud\_f,  
> t3.IS\_DEL i\_f  
> FROM  
> system\_district t1,  
> system\_service\_station t2,  
> system\_fix\_station t3  
> WHERE  
> t1.ID = t2.REGION\_ID  
> AND t1.ID = t3.REGION\_ID"  
> schedule =\> "\* \* \* \* \*"  
> type =\> "system\_district"  
> }  
> }  
> filter {  
> json {  
> source =\> "message"  
> remove\_field =\> ["message"]  
> }  
> aggregate {  
> task\_id =\> "%{id}"  
> code =\> "  
> map['id'] = event.get('id')  
> map['name'] = event.get('name')  
> map['bossCode'] = event.get('bossCode')  
> map['namespaceCode'] = event.get('namespaceCode')  
> map['creator'] = event.get('creator')  
> map['createdDate'] = event.get('createdDate')  
> map['updater'] = event.get('updater')  
> map['updatedDate'] = event.get('updatedDate')  
> map['isDel'] = event.get('isDel')  
> map['systemServiceStation'] ||=  
> map['systemServiceStation'] \<\< {'id' =\> event.get('id\_s'),'name' =\> event.get('name\_s'),'regionId' =\> event.get('rid\_s'),'manger' =\> event.get('m\_s'),'remark' =\> event.get('r\_s'),'bossCode' =\> event.get('b\_s'),'namespaceCode' =\> event.get('n\_s'),'creator' =\> event.get('c\_s'),'createdDate' =\> event.get('cd\_s'),'updater' =\> event.get('u\_s'),'updatedDate' =\> event.get('ud\_s'),'isDel' =\> event.get('i\_s')}  
> map['systemFixStation'] ||=  
> map['systemFixStation'] \<\< {'id' =\> event.get('id\_f'),'name' =\> event.get('name\_f'),'regionId' =\> event.get('rid\_f'),'manger' =\> event.get('m\_f'),'remark' =\> event.get('r\_f'),'bossCode' =\> event.get('b\_f'),'namespaceCode' =\> event.get('n\_f'),'creator' =\> event.get('c\_f'),'createdDate' =\> event.get('cd\_f'),'updater' =\> event.get('u\_f'),'updatedDate' =\> event.get('ud\_f'),'isDel' =\> event.get('i\_f')}  
> event.cancel()  
> "  
> push\_previous\_map\_as\_event =\> true  
> timeout =\> 3  
> }  
> mutate {  
> remove\_field =\> ["@timestamp", "@version", "tags"]  
> remove\_tag =\> ["tags"]  
> }  
> }  
> output {  
> elasticsearch {  
> hosts =\> ["192.168.88.128:9200","192.168.88.129:9200"]  
> action =\> "index"  
> index =\> "system\_district"  
> document\_id=\>"%{id}"  
> manage\_template =\> true  
> template =\> "/usr/local/logstash-6.2.1/template/mappings/system\_district.json"  
> template\_name =\> "system\_district.json"  
> template\_overwrite =\> true  
> }  
> stdout {  
> codec =\> json\_lines  
> }  
> }  
> I would be grateful if anyone could give me some advice.
> 
> ![a](https://us1.discourse-cdn.com/elastic/original/3X/7/9/79133250bb56131f15d1ef51a50de5b2cf2b54fd.png)

---

<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:** [August 7, 2018, 7:19am UTC](https://discuss.elastic.co/t/duplication-of-data-using-the-aggregate-plugin/139284/2 "2018-08-07T07:19:25Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.