文章轉載自：
https://blog.csdn.net/laoyang360/article/details/51747266
引言：
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)
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
[注意啦！注意啦！注意啦！]
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的話，請留言告訴我，不吝賜教！
————————————————
版權聲明：本文為CSDN博主「銘毅天下」的原創文章，遵循CC 4.0 BY-SA版權協議，轉載請附上原文出處鏈接及本聲明。
原文鏈接：https://blog.csdn.net/laoyang360/java/article/details/51747266
