# Loading large table from Mysql to ES via Logstash
**URL:** https://discuss.elastic.co/t/loading-large-table-from-mysql-to-es-via-logstash/323442
**Category:** Logstash
**Created:** [January 18, 2023, 6:07pm UTC](https://discuss.elastic.co/t/loading-large-table-from-mysql-to-es-via-logstash/323442 "2023-01-18T18:07:08Z")
**Posts on this page:** 9
**Page:** 1
### Author:  [@Het\_Desai](https://discuss.elastic.co/u/Het_Desai)
#### Post date: [January 18, 2023, 6:07pm UTC](https://discuss.elastic.co/t/loading-large-table-from-mysql-to-es-via-logstash/323442/1 "2023-01-18T18:07:08Z")
Hello,
I have multiple large tables (each table contains ~15M records and consumes ~70GB of data).
My logstash configuration is as given below.
```auto
input {
jdbc {
jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/database?useSSL=false"
jdbc_user => "root"
jdbc_password => "root"
jdbc_validate_connection => true
jdbc_driver_library => "/Users/admin/Downloads/logstash-6.5.4/logstash-core/lib/mysql-connector-java-6.0.5.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
statement => "select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1"
jdbc_paging_enabled => true
jdbc_page_size => "5000"
}
}
filter {
mutate {
add_field => { "status" => "add"}
}
}
output {
elasticsearch {
index => "index_123"
document_type => "document"
user => "elastic"
password => "pass"
hosts => "localhost:9200"
document_id => "%{id}"
action => "update"
doc_as_upsert => "true"
}
}
```
For each table logstash takes 4-5 days to log data from MySql to ES. Initially it runs fast but as it goes down in the table, it takes longer and longer for each batch of data.
Initially:
```auto
(0.762422s) SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1) AS `t1` LIMIT 5000 OFFSET 0
(0.865474s) SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1) AS `t1` LIMIT 5000 OFFSET 5000
(0.869017s) SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1) AS `t1` LIMIT 5000 OFFSET 10000
```
After few million rows:
```auto
(582.38830s) SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1) AS `t1` LIMIT 5000 OFFSET 9365000
(580.187553s) SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1) AS `t1` LIMIT 5000 OFFSET 9370000
(586.855813s) SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1) AS `t1` LIMIT 5000 OFFSET 9375000
```
I have a use case where I want to log updated columns of a table, frquently.
I have tried **sql\_last\_value** feature with **tracking\_column** and **tracking\_column\_type**.
```auto
input {
jdbc {
jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/database?useSSL=false"
jdbc_user => "root"
jdbc_password => "root"
jdbc_validate_connection => true
jdbc_driver_library => "/Users/admin/Downloads/logstash-6.5.4/logstash-core/lib/mysql-connector-java-6.0.5.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
statement => "select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1 and id > :sql_last_value"
jdbc_paging_enabled => true
jdbc_page_size => "5000"
use_column_value => true
tracking_column => "id"
tracking_column_type => "numeric"
}
}
filter {
mutate {
add_field => { "status" => "add"}
}
}
output {
elasticsearch {
index => "index_123"
document_type => "document"
user => "elastic"
password => "pass"
hosts => "localhost:9200"
document_id => "%{id}"
action => "update"
doc_as_upsert => "true"
}
}
```
But it stays id \> 0 for each query.
```auto
SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1 and id > 0) AS `t1` LIMIT 5000 OFFSET 0
SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1 and id > 0) AS `t1` LIMIT 5000 OFFSET 5000
SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1 and id > 0) AS `t1` LIMIT 5000 OFFSET 10000
```
As per testing I found that last value will get stored and used in next pipeline process.
But to optimize logstash process, I want something like this,
```auto
SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1 and id > 0) AS `t1` LIMIT 5000
SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1 and id > 5000) AS `t1` LIMIT 5000
SELECT * FROM (select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1 and id > 10000) AS `t1` LIMIT 5000
```
Because I think dynamic where clause ( id \> lastId ) would optimize process.
Any help would be highly appriciated !
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [January 18, 2023, 7:11pm UTC](https://discuss.elastic.co/t/loading-large-table-from-mysql-to-es-via-logstash/323442/2 "2023-01-18T19:11:51Z")
> [@Het\_Desai](#):
>
> (580.187553s) SELECT \* FROM (select id, updated\_at, updated\_column1, updated\_column2 from database.table where isUpdated = 1) AS `t1` LIMIT 5000 OFFSET 9370000
I'm not a jdbc user, so there is some speculation embedded here, but I will offer it as a line of investigation for you.
Your basic problem here appears to be that each time you execute the query you are fetching 9.4 million rows from the table and then throwing away almost all of them away. Then you are doing that again for the next 5000 rows. That takes time.
Any solution that involves a huge OFFSET value will be slow -- paging will not work well with this result set.
If 'id' is a key with which you can order all the rows and use as the tracking column, then you may be able to optimize this.
Schedule the input to run once per second. That [means](https://stackoverflow.com/questions/64420859/what-happens-when-a-scheduled-task-didnt-finish-before-next-scheduled-time-star/64432487#64432487) that the input will run the query to fetch a bunch of rows, then quickly start the next iteration.
I don't speak SQL, so I don't know how to write the query, but in English what you want is to do a select with 'ORDER BY id' from the entire table, and from that select 'id \> :sql\_last\_value' with 'LIMIT 5000'. This effectively implements paging in a different way.
Provided that the index fits in the buffer cache that should perform reasonably well. If the index does not fit it will thrash the cache and performance will go to hell in a handbasket.
---
### Author:  [@Het\_Desai](https://discuss.elastic.co/u/Het_Desai)
#### Post date: [January 19, 2023, 3:39pm UTC](https://discuss.elastic.co/t/loading-large-table-from-mysql-to-es-via-logstash/323442/3 "2023-01-19T15:39:26Z")
@Badger thanks for replying 🙂
> [@Badger](#):
>
> Your basic problem here appears to be that each time you execute the query you are fetching 9.4 million rows from the table and then throwing away almost all of them away. Then you are doing that again for the next 5000 rows. That takes time.
Absolutly, that is the behaviour of logstash-jdbc when you give query-statement with paging enabled. If logstash has a feature to assign dynamic value at every query while paging (Just like what it does with offset value), this type of use case would get benefit.
> [@Badger](#):
>
> I don't speak SQL, so I don't know how to write the query, but in English what you want is to do a select with 'ORDER BY id' from the entire table, and from that select 'id \> :sql\_last\_value' with 'LIMIT 5000'. This effectively implements paging in a different way.
I tried the suggested solution and it's a nice solution to keep it logging latest data. It can be done by configuring jdbc-input like this (You may not need limit in this method):
```auto
jdbc {
jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/database?useSSL=false"
jdbc_user => "root"
jdbc_password => "root"
jdbc_validate_connection => true
jdbc_driver_library => "/Users/admin/Downloads/logstash-6.5.4/logstash-core/lib/mysql-connector-java-6.0.5.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
statement => "select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1 and id > :sql_last_value LIMIT 5000"
schedule => "* * * * *"
use_column_value => true
tracking_column => "id"
tracking_column_type => "numeric"
last_run_metadata_path => "./sql_last_value.yml"
}
```
But If someone has a data pipeline processes (like me) including logstash being one of them then it's not the solution. It has following possible issues.
1. Scheduling can't be done at every second. Minimum cron interval can be 60s only.
2. Second thing is that it's cron job that will trigger process at every interval. So you don't know exactly when logstash process is completed after which you may want to kill it and move onto next process.
---
### Author:  [@Het\_Desai](https://discuss.elastic.co/u/Het_Desai)
#### Post date: [January 19, 2023, 3:56pm UTC](https://discuss.elastic.co/t/loading-large-table-from-mysql-to-es-via-logstash/323442/4 "2023-01-19T15:56:07Z")
I have implemented a solution, If someone faces the situation just like me.
You can invoke logstash for batches of data, synchronously.
```auto
jdbc {
jdbc_connection_string => "jdbc:mysql://127.0.0.1:3306/database?useSSL=false"
jdbc_user => "root"
jdbc_password => "root"
jdbc_validate_connection => true
jdbc_driver_library => "/Users/admin/Downloads/logstash-6.5.4/logstash-core/lib/mysql-connector-java-6.0.5.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
statement => "select id, updated_at, updated_column1, updated_column2 from database.table where isUpdated = 1 and id >= firstIdOfBatch and id <= lastIdOfBatch"
jdbc_paging_enabled => true
jdbc_page_size => "5000"
}
```
The key being here is where clause:
> where isUpdated = 1 and id \>= firstIdOfBatch and id \<= lastIdOfBatch;
Here you may need to decide optimum size of data-batch and jdbc\_page\_size.
I have kept jdbc\_page\_size: 5000 and data-batch is 0.2M.
The problem with this solution is the cost of time and CPU of invoking logstash multiple times. But for now I have saved my time with this solution compared to original setting. Now instead of taking 4-5 days for one table, I am able to complete logstash process in 3-4 hours.
---
### Author:  [@Het\_Desai](https://discuss.elastic.co/u/Het_Desai)
#### Post date: [January 19, 2023, 4:03pm UTC](https://discuss.elastic.co/t/loading-large-table-from-mysql-to-es-via-logstash/323442/5 "2023-01-19T16:03:11Z")
Still the best way would be to be able to have query parameter dynamically like this:
```auto
SELECT * FROM (select [COLUMNS] from database.table where id > 0) AS `t1` LIMIT 5000
SELECT * FROM (select [COLUMNS] from database.table where id > 5000) AS `t1` LIMIT 5000
SELECT * FROM (select [COLUMNS] from database.table where id > 10000) AS `t1` LIMIT 5000
```
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [January 19, 2023, 5:50pm UTC](https://discuss.elastic.co/t/loading-large-table-from-mysql-to-es-via-logstash/323442/6 "2023-01-19T17:50:23Z")
> [@Het\_Desai](#):
>
> - Scheduling can't be done at every second. Minimum cron interval can be 60s only.
> - Second thing is that it's cron job that will trigger process at every interval. So you don't know exactly when logstash process is completed after which you may want to kill it and move onto next process.
Not so. The schedule option on the jdbc input uses the rufus scheduler, and when parsing cronlines if there are six fields the first one is treated as seconds. A single logstash process will execute the query as often as the cronline tells it to, the process is never complete.
---
### Author:  [@Het\_Desai](https://discuss.elastic.co/u/Het_Desai)
#### Post date: [January 20, 2023, 3:45pm UTC](https://discuss.elastic.co/t/loading-large-table-from-mysql-to-es-via-logstash/323442/7 "2023-01-20T15:45:14Z")
> [@Badger](#):
>
> Not so. The schedule option on the jdbc input uses the rufus scheduler, and when parsing cronlines if there are six fields the first one is treated as seconds
Ohh, my bad. I didn't check if rufus schedular offers this facility.
---
### Author:  [@elasticforme](https://discuss.elastic.co/u/elasticforme)
#### Post date: [January 20, 2023, 7:07pm UTC](https://discuss.elastic.co/t/loading-large-table-from-mysql-to-es-via-logstash/323442/8 "2023-01-20T19:07:53Z")
Hi welcome to the forum,
Without going in to lot more detail. I have done this process. somethink like this.
database has multimillion records and this is what we do initially
for first load of data we use some period (three month in my case) until we are up-to-date.
We run this logstash from command line
SQL we use.
select x,y,z from database where update\_time \> x-date and update\_time \< y-date ( three month, then next three month and so on)
once all data are pulled till today then logstash goes in pipleline and ingest data every 30 min
select x,y,z from database where update\_time \> sysdate-interval '30' minute
In this process we are not relying on sql\_last\_value or tracking\_column. But all happens by updated\_time column in database table. because every record when updates also update this column.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [February 17, 2023, 7:08pm UTC](https://discuss.elastic.co/t/loading-large-table-from-mysql-to-es-via-logstash/323442/9 "2023-02-17T19:08:07Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.