# Logstash synchronize mysql data to elastic

**URL:** <https://discuss.elastic.co/t/logstash-synchronize-mysql-data-to-elastic/168976>\
**Category:** Logstash\
**Created:** [February 19, 2019, 9:26am UTC](https://discuss.elastic.co/t/logstash-synchronize-mysql-data-to-elastic/168976 "2019-02-19T09:26:37Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![Zoomtoo](https://avatars.discourse-cdn.com/v4/letter/z/439d5e/32.png) [@Zoomtoo](https://discuss.elastic.co/u/Zoomtoo)\
**Post date:** [February 19, 2019, 9:26am UTC](https://discuss.elastic.co/t/logstash-synchronize-mysql-data-to-elastic/168976/1 "2019-02-19T09:26:37Z")

</div>

Hello experts:  
My software version (es-6.5.1, logstash-6.5.1, mysql5.7)  
Recently, I encountered a problem when using logstash to synchronize mysql data to elastic. Synchronize data from the same library in mysql. After multi-table synchronization to es, the time difference of the date\_tine type field in mysql is different, such as table. The add\_time in 1 will increase by 6h in es, and the update\_time in table 2 will increase by 14h in es. There is another case where I synchronize the same data in one table, the time difference of the first execution synchronization and the second execution synchronization. The time difference is inconsistent. What is going on? Is there any instability factor in logstash? I don’t have a clue at present, please advise.  
Below I list a few of my logstash synchronization profiles：  
1.conf  
input{  
stdin{  
}  
jdbc {  
# 数据库  
jdbc\_connection\_string =\> "jdbc:mysql://xxx:xxx:xxx:xxx:93306/test?useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull"  
jdbc\_user =\> "test"  
jdbc\_password =\> "1"  
# mysql驱动解压的位置  
jdbc\_driver\_library =\> "/elk/logstash-6.5.1/logstash-core/lib/jars/mysql-connector-java-8.0.13.jar"  
# mysql驱动类  
jdbc\_driver\_class =\> "com.mysql.jdbc.Driver"  
jdbc\_paging\_enabled =\> "true"  
jdbc\_page\_size =\> "50000"  
#use\_column\_value =\> true  
#tracking\_column =\> id  
record\_last\_run =\> true  
#last\_run\_metadata\_path =\> "/elk/logstash-6.5.1/bin/config\_mysql/flag/axh\_flag/axh\_user.txt"  
#可以使用sql文件的方式  
#statement\_filepath =\> "/home/elk1/software/logstash-6.5.3/bin/config-mysql/axh\_user.sql"  
#要同步的表  
statement =\> "select id,uid,add\_time from axh\_active\_user where add\_time like '2019-02-18%'"  
schedule =\> "04 17 \* \* _"  
 #索引type  
type =\> "axh\_active\_user"  
 #索引设置时区  
# jdbc\_default\_timezone =\> "Asia/Shanghai"  
}  
}  
filter {  
ruby {  
code =\>"event.set('new\_date', event.get('@timestamp').time.localtime + 8_60\*60)"  
}  
ruby {  
code =\> "event.set('@timestamp',event.get('new\_date'))"  
}  
mutate {  
remove\_field =\> ["new\_date"]  
}  
date {  
match =\> ["add\_time", "UNIX\_MS"]  
target =\> "@timestamp"  
}  
}  
output {  
elasticsearch {  
hosts =\> "xxx:xxx:xxx:xxx:xx"  
#索引名称  
index =\> "test\_time\_new"  
document\_id =\> "%{id}"  
user =\> "elastic"  
password =\> "123"  
#输出的索引type此处表示上面的tag  
"document\_type" =\> "%{type}"  
}  
stdout {  
codec =\> rubydebug  
}  
}

2.conf

input {  
stdin{  
}  
jdbc {  
# 数据库  
jdbc\_connection\_string =\> "jdbc:mysql://xxx:xxx:xxx:xxx:93306/test?useUnicode=true&characterEncoding=utf-8&zeroDateTimeBehavior=convertToNull"  
jdbc\_user =\> "test"  
jdbc\_password =\> "1"  
# mysql驱动解压的位置  
jdbc\_driver\_library =\> "/elk/logstash-6.5.1/logstash-core/lib/jars/mysql-connector-java-8.0.13.jar"  
# mysql驱动类  
jdbc\_driver\_class =\> "com.mysql.jdbc.Driver"  
jdbc\_paging\_enabled =\> "true"  
jdbc\_page\_size =\> "50000"  
#use\_column\_value =\> true  
#tracking\_column =\> id  
record\_last\_run =\> true  
#last\_run\_metadata\_path =\> "/elk/logstash-6.5.1/bin/config\_mysql/flag/axh\_flag/axh\_user.txt"  
#可以使用sql文件的方式  
#statement\_filepath =\> "/home/elk1/software/logstash-6.5.3/bin/config-mysql/axh\_user.sql"  
#要同步的表  
statement =\> "select id, reg\_time,date\_add(reg\_time,INTERVAL -14 HOUR) as real\_reg\_time,tg\_status_1 as tg\_status,mobile,tg\_id,is\_vip_1 as is\_vip,date\_add(vip\_start\_time,INTERVAL -14 HOUR) as vip\_start\_time,is\_login\*1 as is\_login,date\_add(last\_login\_time,INTERVAL -14 HOUR) as last\_login\_time from axh\_user where id\>=1432356"  
schedule =\> "23 14 \* \* \*"  
#索引type  
type =\> "axh\_user"  
#索引设置时区  
# jdbc\_default\_timezone =\> "Asia/Shanghai"  
}  
}  
filter {  
ruby {  
code =\> "event.timestamp.time.localtime"  
}  
}  
output {  
elasticsearch {  
hosts =\> "xxx:xxx:xxx:xxx:xx"  
#索引名称  
index =\> "test\_time"  
document\_id =\> "%{id}"  
user =\> "elastic"  
password =\> "123"  
#输出的索引type此处表示上面的tag  
"document\_type" =\> "%{type}"  
}  
stdout {  
codec =\> json\_lines  
}  
}

---

<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:** [March 19, 2019, 9:26am UTC](https://discuss.elastic.co/t/logstash-synchronize-mysql-data-to-elastic/168976/2 "2019-03-19T09:26:58Z")

</div>

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