logstash-input-jdbc实现mysql 与elasticsearch实时同步深刻详解

elasticsearch 的出现使得咱们的存储、检索数据更快捷、方便。但不少状况下，咱们的需求是：如今的数据存储在MySQL、Oracle等关系型传统数据库中，如何尽可能不改变原有数据库表结构，将这些数据的insert,update,delete操做结果实时同步到elasticsearch(简称ES)呢？
本文基于以上需求点展开实战讨论。

1.对delete操做的实时同步泼冷水

到目前为止，全部google,stackoverflow,elastic.co,github上面搜索的插件和实时同步的信息，告诉咱们：目前同步delete尚未好的解决方案。
折中的解决方案以下：
方案探讨：https://discuss.elastic.co/t/delete-elasticsearch-document-with-logstash-jdbc-input/47490/9
http://stackoverflow.com/questions/34477095/elasticsearch-replication-of-other-system-data/34477639#34477639

方案一,

在原有的mysql数据库表中，新增一个字段status, 默认值为ok，若是要删除数据，实则用update操做，status改成deleted.
这样，就能同步到es中。es中以status状态值区分该行数据是否存在。deleted表明已删除，ok表明正常。

方案二，

使用Go elasticsearch 插件实现同步，如：。可是我实操发现，该插件不稳定，bug较多。我也给源码做者提出了bug。
Bug详见：https://github.com/siddontang/go-mysql-elasticsearch/issues/46

关于删除操做的最终讨论解决方案（截止2016年6月24日）：

http://stackoverflow.com/questions/35813923/sync-postgresql-data-with-elasticsearch/35823497#35823497

首先，软件删除而非物理删除数据，新增一个 flag 列，标识记录是否已经被删除，这样，相同的记录也会存在于Elasticsearch。能够执行简单的term查询操做,检索出已经删除的数据信息。
其次，若须要执行cleanup清理数据操做（物理删除），只须要在数据库和ES中同时删除掉标记位deleted的记录便可。如：mysql执行：delete from cc where cc.flag=’deleted’; ES一样执行对应删除操做。

2.如何使用 插件实现insert,update 的同步更新操做？

个人上一篇博文：http://blog.csdn.net/laoyang360/article/details/51694519 作了些许探讨。
除了上篇文章提到的三个插件，这里推荐试用过比较好用的logstash的一款插件，名称为： logstash-input-jdbc

3.如何安装logstash-input-jdbc插件？

参考：http://blog.csdn.net/yeyuma/article/details/50240595#quote
网友博文已经介绍很详细，再也不赘述。
基本到这一步：

cd /opt/logstash/

sudo bin/plugin install logstash-input-jdbc

到此，基本就能成功。若不能请留言。

4,如何实现实时同步？

4.1 前提：mysql存在的数据库及表

数据库名为:test
test下表名为:cc
表中数据为：

mysql> use test;
Reading table information for completion of table and column names You can turn off this feature to get a quicker startup with -A Database changed mysql> select * from cc; +----+--------------------+---------+---------------------+ | id | name | status | modified_at | +----+--------------------+---------+---------------------+ | 1 | laoyang360 | ok | 0000-00-00 00:00:00 | | 2 | test002 | ok | 2016-06-23 06:16:42 | | 3 | dllaoyang | ok | 0000-00-00 00:00:00 | | 4 | huawei | ok | 0000-00-00 00:00:00 | | 5 | jdbc_test_update08 | ok | 0000-00-00 00:00:00 | | 7 | test7 | ok | 0000-00-00 00:00:00 | | 8 | test008 | ok | 0000-00-00 00:00:00 | | 9 | test9 | ok | 0000-00-00 00:00:00 | | 10 | test10 | deleted | 0000-00-00 00:00:00 | +----+--------------------+---------+---------------------+ 9 rows in set (0.01 sec)

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20

4.2 须要两个文件：1）jdbc.conf; 2)jdbc.sql.

[root@5b9dbaaa148a logstash_jdbc_test]# cat jdbc.conf input { stdin { } jdbc { # mysql jdbc connection string to our backup databse 后面的test对应mysql中的test数据库 jdbc_connection_string => "jdbc:mysql://192.168.1.1:3306/test" # the user we wish to excute our statement as jdbc_user => "root" jdbc_password => "******" # the path to our downloaded jdbc driver jdbc_driver_library => "/elasticsearch-jdbc-2.3.2.0/lib/mysql-connector-java-5.1.38.jar" # the name of the driver class for mysql jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_paging_enabled => "true" jdbc_page_size => "50000" #如下对应着要执行的sql的绝对路径。 statement_filepath => "/usr/local/logstash/bin/logstash_jdbc_test/jdbc.sql" #定时字段 各字段含义（由左至右）分、时、天、月、年，所有为*默认含义为每分钟都更新（测试结果，不一样的话请留言指出） schedule => "* * * * *" #设定ES索引类型 type => "cc_type" } } filter { json { source => "message" remove_field => ["message"] } } output { elasticsearch { #ESIP地址与端口 hosts => "192.168.1.1:9200" #ES索引名称（本身定义的） index => "cc_index" #自增ID编号 document_id => "%{id}" } stdout { #以JSON格式输出 codec => json_lines } } #要执行的sql语句。 选择哪些信息同步到ES中。 [root@5b9dbaaa148a logstash_jdbc_test]# cat jdbc.sql select * from where cc.modified_at > :sql_last_value

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38
• 39
• 40
• 41
• 42
• 43
• 44
• 45
• 46
• 47
• 48
• 49
• 50
• 51
• 52
• 53
• 54

[注意啦！注意啦！注意啦！]

cc.modified_at, 这个modified_at是我本身定义的更改时间字段，默认值default是now()当前时间。
而 :sql_last_value若是input里面use_column_value => true， 即若是设置为true的话，能够是咱们设定的字段的上一次的值。
默认 use_column_value => false， 这样 :sql_last_value为上一次更新的最后时刻值。
也就是说，对于新增的值，才会更新。这样就实现了增量更新的目的。

有童鞋问，如何全量更新呢？ 答案：就是去掉where子句便可。

步骤1：

在logstash的bin路径下新建文件夹logstash_jdbc_test，并将上两个文件 1）jdbc.conf，2)jdbc.sql.模板拷贝到里面。

步骤2：

按照本身的mysql地址、es地址、创建的索引名称、类型名称修改conf，以及要同步内容修改sql。

步骤3：

执行logstash, 以下：
[root@5b9dbaaa148a plugins]# ./logstash -f ./logstash_jdbc_test/jdbc.conf

步骤4：

验证同步是否成功。
能够经过： 以下图所示：

5,注意事项

若是你要测试go-mysql-elasticsearch可能会遇到下面三个Bug及解决方案以下：

【Bug1】

How to Setting The Binary Log Format
http://dev.mysql.com/doc/refman/5.7/en/binary-log-setting.html

【Bug2】

what is inner http status address
https://github.com/siddontang/go-mysql-elasticsearch/issues/11

【Bug3】

[2016/06/23 10:19:38] canal.go:146 [Error] canal start sync binlog err: ERROR 1236 (HY000): Misconfigured master - server id was not set
http://dba.stackexchange.com/questions/76089/error-1236-from-master-after-restored-replication

6,小结

实操发现： logstash-input-jdbc 能较好的实现mysql的insert、update的操做的增量、全量数据同步更新到ES。
但delete操做的实时同步没有很好的解决方案，若是你有，且都测试ok的话，请留言告诉我，不吝赐教！

2016-6-23 pm22:42 思于家中床前

做者：铭毅天下
转载请标明出处，原文地址：http://blog.csdn.net/laoyang360/article/details/51747266