logstash-input-jdbc實現mysql 與elasticsearch實時同步深入詳解
引言:
elasticsearch 的出現使得我們的儲存、檢索資料更快捷、方便。但很多情況下,我們的需求是:現在的資料儲存在mysql、oracle等關係型傳統資料庫中,如何儘量不改變原有資料庫表結構,將這些資料的insert,update,delete操作結果實時同步到elasticsearch(簡稱ES)呢?
本文基於以上需求點展開實戰討論。
1.對delete操作的實時同步潑冷水
到目前為止,所有google,stackoverflow,elastic.co,github上面搜尋的外掛和實時同步的資訊,告訴我們:目前同步delete還沒有好的解決方案。
折中的解決方案如下:
方案探討:
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.
[[email protected]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中。
[[email protected]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, 如下:
[[email protected] 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的話,請留言告訴我,不吝賜教!
——————————————————————————————————
更多ES相關實戰乾貨經驗分享,請掃描下方【銘毅天下】微信公眾號二維碼關注。
(每週至少更新一篇!)
和你一起,死磕Elasticsearch!
——————————————————————————————————
2016-6-23 pm22:42 思於家中床前
作者:銘毅天下
轉載請標明出處,原文地址:http://blog.csdn.net/laoyang360/article/details/51747266
如果感覺本文對您有幫助,請點選‘頂’支援一下,您的支援是我堅持寫作最大的動力,謝謝!
再分享一下我老師大神的人工智慧教程吧。零基礎!通俗易懂!風趣幽默!希望你也加入到我們人工智慧的隊伍中來!http://www.captainbed.net