# Mysql to Elasticsearch Sync with Logstash & input JDBC plugin

**URL:** <https://discuss.elastic.co/t/mysql-to-elasticsearch-sync-with-logstash-input-jdbc-plugin/159917>\
**Category:** Logstash\
**Created:** [December 7, 2018, 11:22am UTC](https://discuss.elastic.co/t/mysql-to-elasticsearch-sync-with-logstash-input-jdbc-plugin/159917 "2018-12-07T11:22:41Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![XMM](https://avatars.discourse-cdn.com/v4/letter/x/a3d4f5/32.png) [@XMM](https://discuss.elastic.co/u/XMM)\
**Post date:** [December 7, 2018, 11:22am UTC](https://discuss.elastic.co/t/mysql-to-elasticsearch-sync-with-logstash-input-jdbc-plugin/159917/1 "2018-12-07T11:22:42Z")

</div>

Hello,  
i am trying to synchronize my MySql database to ElasticSearch with usage of Logstash and JDBC input plugin. As far I have working "dummy" prototype to sync entire Sql table row to single document. My challenge comes now because I am trying to achieve next:  
Customer has many addresses  
Customer has many contacts...

I would like to have document which looks like this (focused on contact entity but i believe the is the same behaviour also for multiple has many relations..)

Customer {  
id:1,  
name: "testq",  
lastname: "lastname",  
contacts: {  
{  
id:1,  
[contact:"testemail@email.com](mailto:contact:%22testemail@email.com)"},  
{  
id:2,  
contact:"123456763545"}  
},  
addresses:{  
{addressdata1},{addressdata2}  
}  
....  
}

Currently i figure out how to sync customer entity and keep it updated (to update existing one).  
I try to achieve the same for "contacts" and "addresses" and other entities..

With help of other topic: [https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html#plugins-filters-aggregate-usecases](https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html#plugins-filters-aggregate-usecases)

I manage to get to this code:  
filter {  
aggregate {  
task\_id =\> "%{cust\_id}\_customer"  
code =\> "  
map['id'] = event.get('cust\_id')  
map['firstname'] = event.get('firstname')  
map['lastname'] = event.get('lastname')  
map['username'] = event.get('username')  
map['contacts'] ||=   
map['contacts'] \<\< {  
'customer\_contact\_id' =\> event.get('customer\_contact\_id'),  
'contact' =\> event.get('contact'),  
'used\_for' =\> event.get('used\_for'),  
'type' =\> event.get('type'),  
'status' =\> event.get('status'),  
'created' =\> event.get('created'),  
'modified' =\> event.get('modified')  
}  
event.cancel()  
"  
push\_previous\_map\_as\_event =\> true  
timeout =\> 5000  
}  
}

My questions will be next:

1. Is there a way not to specify all fields on customer level to sync? (fistname, lastname) but just to append new contacts, addresses other... array into customer model?
2. Is there a way to update contact data by their "customer.contacts.id" like on customer level? (currently all records are delted and reinserted)?
3. Is there a way not to specify each contact fields bust just create arrays..{} and append them to contacts?
4. I face that not every time all records are synched. One customer has 4 contacts but in most synches min 2 are synched. any purposal why this is happening?

Mysql query i run is:  
SELECT Customer._,Customer.id as cust\_id,CustomerContact._,CustomerContact.id as customer\_contact\_id  
FROM customers\_\_customers as Customer  
LEFT JOIN customers\_\_customer\_contacts as CustomerContact ON CustomerContact.customer\_id = Customer.id  
ORDER by Customer.id asc

which return all contacts for customer.

I am also thinking about having one document with uuid of customer id and then bunch of arrays  
{  
id:1  
customer:{  
id:1  
name: testq  
....},  
contacts:{  
{ id:,  
[contact:"testemail@email.com](mailto:contact:%22testemail@email.com)"...},  
...},  
addresses:{}  
}

Which is more optimal form for update and maintenance and especially for further ES searches?

Thanks for reply.

---

<div class="post-metadata">

**Author:** ![guyboertje](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/guyboertje/32/31592_2.png) [@guyboertje](https://discuss.elastic.co/u/guyboertje)\
**Post date:** [December 7, 2018, 11:30am UTC](https://discuss.elastic.co/t/mysql-to-elasticsearch-sync-with-logstash-input-jdbc-plugin/159917/2 "2018-12-07T11:30:46Z")

</div>

Typically you would use jdbc\_streaming or jdbc\_static to achieve the "JOIN".

So:  
jdbc input (fetch customers) -\> some filters -\> jdbc\_streaming(use customer\_contact\_id to query contacts table) -\> jdbc\_streaming(use customer\_address\_id to query addresses) -\> ES

this is denormalising.

---

<div class="post-metadata">

**Author:** ![XMM](https://avatars.discourse-cdn.com/v4/letter/x/a3d4f5/32.png) [@XMM](https://discuss.elastic.co/u/XMM)\
**Post date:** [December 11, 2018, 7:27am UTC](https://discuss.elastic.co/t/mysql-to-elasticsearch-sync-with-logstash-input-jdbc-plugin/159917/3 "2018-12-11T07:27:56Z")

</div>

Hello,  
i solved my case by using your purposed plugin jdbc\_streaming filter and next config:  
filter {  
jdbc\_streaming {  
id =\> "contacts"  
jdbc\_driver\_library =\> "/usr/share/java/mysql.jar"  
jdbc\_driver\_class =\> "com.mysql.jdbc.Driver"  
jdbc\_connection\_string =\> "jdbc:mysql://localhost:3306/23123"  
jdbc\_user =\> "123123"  
jdbc\_password =\> "123123"  
statement =\> "SELECT \* FROM customers\_\_customer\_contacts as CustomerContact WHERE customer\_id = :id"  
parameters =\> { "id" =\> "id"}  
target =\> "contacts"  
}  
jdbc\_streaming {  
id =\> "addresses"  
jdbc\_driver\_library =\> "/usr/share/java/mysql.jar"  
jdbc\_driver\_class =\> "com.mysql.jdbc.Driver"  
jdbc\_connection\_string =\> "jdbc:mysql://localhost:3306/13"  
jdbc\_user =\> "123123"  
jdbc\_password =\> "123123"  
statement =\> "SELECT \* FROM customers\_\_customer\_addresses WHERE customer\_id = :id"  
parameters =\> { "id" =\> "id"}  
target =\> "addresses"  
}  
}

Thanks for pointing me in right direction! 🙂

---

<div class="post-metadata">

**Author:** ![guyboertje](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/guyboertje/32/31592_2.png) [@guyboertje](https://discuss.elastic.co/u/guyboertje)\
**Post date:** [December 11, 2018, 9:50am UTC](https://discuss.elastic.co/t/mysql-to-elasticsearch-sync-with-logstash-input-jdbc-plugin/159917/4 "2018-12-11T09:50:36Z")

</div>

Good to know.

Experiment with the cache settings so that more reference (customers/contacts) are in memory while running but bear in mind volatility - how often does is a customer or contact updated etc.

---

<div class="post-metadata">

**Author:** ![XMM](https://avatars.discourse-cdn.com/v4/letter/x/a3d4f5/32.png) [@XMM](https://discuss.elastic.co/u/XMM)\
**Post date:** [December 13, 2018, 12:55pm UTC](https://discuss.elastic.co/t/mysql-to-elasticsearch-sync-with-logstash-input-jdbc-plugin/159917/5 "2018-12-13T12:55:10Z")

</div>

One more observation. As target type of setting is "string". May i somehow add 3 or more levels of array?

customer

- order  
-order\_item  
-product data

i have similar config for orders, and i want to enrich it with order\_items data?  
target =\> "order.items"

Thanks in advance.

---

<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:** [January 10, 2019, 12:55pm UTC](https://discuss.elastic.co/t/mysql-to-elasticsearch-sync-with-logstash-input-jdbc-plugin/159917/6 "2019-01-10T12:55:14Z")

</div>

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