logstash-input-jdbc實現mysql 與elasticsearch實時同步(ES與關系型數據庫同步)


引言:

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插件?

【注意啦,注意啦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 
表中數據為:


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

4.2 需要兩個文件:1)jdbc.conf; 2)jdbc.sql.


   
   
  
  
          
  1. [root@ 5b9dbaaa148a logstash_jdbc_test] # cat jdbc.conf
  2. input {
  3. stdin {
  4. }
  5. jdbc {
  6. # mysql jdbc connection string to our backup databse 后面的test對應mysql中的test數據庫
  7. jdbc_connection_string => "jdbc:mysql://192.168.1.1:3306/test"
  8. # the user we wish to excute our statement as
  9. jdbc_user => "root"
  10. jdbc_password => "******"
  11. # the path to our downloaded jdbc driver
  12. jdbc_driver_library => "/elasticsearch-jdbc-2.3.2.0/lib/mysql-connector-java-5.1.38.jar"
  13. # the name of the driver class for mysql
  14. jdbc_driver_class => "com.mysql.jdbc.Driver"
  15. jdbc_paging_enabled => "true"
  16. jdbc_page_size => "50000"
  17. #以下對應着要執行的sql的絕對路徑。
  18. statement_filepath => "/usr/local/logstash/bin/logstash_jdbc_test/jdbc.sql"
  19. #定時字段 各字段含義(由左至右)分、時、天、月、年,全部為*默認含義為每分鍾都更新(測試結果,不同的話請留言指出)
  20. schedule => "* * * * *"
  21. #設定ES索引類型
  22. type => "cc_type"
  23. }
  24. }
  25. filter {
  26. json {
  27. source => "message"
  28. remove_field => [ "message"]
  29. }
  30. }
  31. output {
  32. elasticsearch {
  33. #ESIP地址與端口
  34. hosts => "192.168.1.1:9200"
  35. #ES索引名稱(自己定義的)
  36. index => "cc_index"
  37. #自增ID編號
  38. document_id => "%{id}"
  39. }
  40. stdout {
  41. #以JSON格式輸出
  42. codec => json_lines
  43. }
  44. }
  45. #要執行的sql語句。
  46. 選擇哪些信息同步到ES中。
  47. [root@ 5b9dbaaa148a logstash_jdbc_test] # cat jdbc.sql
  48. select
  49. *
  50. from
  51. 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


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM