引言：
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日）：
首先，软件删除而非物理删除数据，新增一个 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插件？
【注意啦，注意啦20170920】：logstash5.X开始，已经至少集成了logstash-input-jdbc插件。所以，你如果使用的是logstash5.X，可以不必再安装，可以直接跳过这一步。
参考：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)
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
[注意啦！注意啦！注意啦！]
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