logstash-input-jdbc實現mysql 與elasticsearch實時同步深刻詳解

時間 2019-12-02

標籤 logstash input jdbc 實現 mysql elasticsearch 實時同步深刻詳解欄目日誌分析简体版

原文原文鏈接

elasticsearch 的出現使得咱們的存儲、檢索數據更快捷、方便。但不少狀況下，咱們的需求是：如今的數據存儲在MySQL、Oracle等關係型傳統數據庫中，如何儘可能不改變原有數據庫表結構，將這些數據的insert,update,delete操做結果實時同步到elasticsearch(簡稱ES)呢？
本文基於以上需求點展開實戰討論。php

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#34477639html

方案一,

在原有的mysql數據庫表中，新增一個字段status, 默認值爲ok，若是要刪除數據，實則用update操做，status改成deleted.
這樣，就能同步到es中。es中以status狀態值區分該行數據是否存在。deleted表明已刪除，ok表明正常。java

方案二，

使用Go elasticsearch 插件實現同步，如：。可是我實操發現，該插件不穩定，bug較多。我也給源碼做者提出了bug。
Bug詳見：https://github.com/siddontang/go-mysql-elasticsearch/issues/46mysql

關於刪除操做的最終討論解決方案（截止2016年6月24日）：

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

首先，軟件刪除而非物理刪除數據，新增一個 flag 列，標識記錄是否已經被刪除，這樣，相同的記錄也會存在於Elasticsearch。能夠執行簡單的term查詢操做,檢索出已經刪除的數據信息。
其次，若須要執行cleanup清理數據操做（物理刪除），只須要在數據庫和ES中同時刪除掉標記位deleted的記錄便可。如：mysql執行：delete from cc where cc.flag=’deleted’; ES一樣執行對應刪除操做。github

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

個人上一篇博文：http://blog.csdn.net/laoyang360/article/details/51694519 作了些許探討。
除了上篇文章提到的三個插件，這裏推薦試用過比較好用的logstash的一款插件，名稱爲： logstash-input-jdbcsql

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

參考：http://blog.csdn.net/yeyuma/article/details/50240595#quote
網友博文已經介紹很詳細，再也不贅述。
基本到這一步：數據庫

cd /opt/logstash/json

sudo bin/plugin install logstash-input-jdbcoracle

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

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子句便可。